Fix for fields using db_column

.github/workflows/tests.yml

@@ -47,6 +47,9 @@ jobs:
         exclude:
         - {python: '3.9', django: 'Django~=5.1.0'}
         - {python: '3.9', django: 'Django~=5.2.0'}
+        - {python: '3.9', django: 'Django~=6.0.0'}
+        - {python: '3.10', django: 'Django~=6.0.0'}
+        - {python: '3.11', django: 'Django~=6.0.0'}
     env:
       allowed_python_failure: '3.14'
     services:

django_cte/cte.py

@@ -167,11 +167,20 @@ def queryset(self):
         qs.query = query
         return qs
 
-    def _resolve_ref(self, name):
+    def _resolve_ref(self, column):
+        name = column.name
+        ref = self.query.resolve_ref(name)
+        if ref is column or column in ref.get_source_expressions():
+            raise ValueError(f"Circular reference: {column} = {ref}")
+
+        if name in self.query.annotations:
+            return CTEColumnRef(name, self.name, ref.output_field)
+
         selected = getattr(self.query, "selected", None)
         if selected and name in selected and name not in self.query.annotations:
-            return Ref(name, self.query.resolve_ref(name))
-        return self.query.resolve_ref(name)
+            return Ref(name, ref)
+
+        return ref
 
     def resolve_expression(self, *args, **kw):
         if self.query is None:

django_cte/meta.py

@@ -2,6 +2,12 @@
 
 from django.db.models.expressions import Col, Expression
 
+try:
+    from django.db.models.expressions import ColPairs as _ColPairs
+except ImportError:
+    class _ColPairs:
+        pass
+
 
 class CTEColumns:
 
@@ -35,9 +41,11 @@ def _ref(self):
                 "Hint: use ExpressionWrapper({cte}.col.{name}, "
                 "output_field=...)".format(cte=self._cte.name, name=self.name)
             )
-        ref = self._cte._resolve_ref(self.name)
-        if ref is self or self in ref.get_source_expressions():
-            raise ValueError("Circular reference: {} = {}".format(self, ref))
+
+        ref = self._cte._resolve_ref(self)
+        if isinstance(ref, _ColPairs):
+            raise ValueError("Cannot reference column pairs directly")
+
         return ref
 
     @property
@@ -58,10 +66,7 @@ def output_field(self):
     def as_sql(self, compiler, connection):
         qn = compiler.quote_name_unless_alias
         ref = self._ref
-        if isinstance(ref, Col) and self.name == "pk":
-            column = ref.target.column
-        else:
-            column = self.name
+        column = ref.target.column if isinstance(ref, Col) else self.name
         return "%s.%s" % (qn(self.table_alias), qn(column)), []
 
     def relabeled_clone(self, relabels):

django_cte/raw.py

@@ -27,6 +27,8 @@ def quote_name_unless_alias(self, name):
 
     class raw_cte_queryset:
         class query:
+            annotations = {}
+
             @staticmethod
             def get_compiler(connection, *, elide_empty=None):
                 return raw_cte_compiler(connection)

pyproject.toml

@@ -28,6 +28,7 @@ classifiers = [
     'Framework :: Django :: 5',
     'Framework :: Django :: 5.1',
     'Framework :: Django :: 5.2',
+    'Framework :: Django :: 6.0',
     'Topic :: Software Development :: Libraries :: Python Modules',
 ]
 dependencies = ["django"]
@@ -51,3 +52,26 @@ name = "django_cte"
 
 [tool.distutils.bdist_wheel]
 universal = true
+
+[tool.tox]
+requires = ["tox>=4.43"]
+env_list = [
+    {product = [{prefix = "py3", start = 9, stop = 12}, ["django42"]]},
+    {product = [{prefix = "py3", start = 10, stop = 12}, ["django50"]]},
+    {product = [{prefix = "py3", start = 10, stop = 13}, ["django51"]]},
+    {product = [{prefix = "py3", start = 10, stop = 14}, ["django52"]]},
+    {product = [{prefix = "py3", start = 12, stop = 14}, ["django60", "djangomain"]]},
+]
+
+[tool.tox.env_run_base]
+default_base_python = "python3"
+commands = [["pytest", {replace = "posargs", default = [], extend = true}]]
+dependency_groups = ["dev"]
+deps = [
+  {replace = "if", condition = "factor.django42", then = ["Django>=4.2,<5.0"], extend = true},
+  {replace = "if", condition = "factor.django50", then = ["Django>=5.0,<5.1"], extend = true},
+  {replace = "if", condition = "factor.django51", then = ["Django>=5.1,<5.2"], extend = true},
+  {replace = "if", condition = "factor.django52", then = ["Django>=5.2,<6.0"], extend = true},
+  {replace = "if", condition = "factor.django60", then = ["Django>=6.0,<6.1"], extend = true},
+  {replace = "if", condition = "factor.djangomain", then = ["https://github.com/django/django/archive/main.tar.gz"], extend = true},
+]

tests/django_setup.py

@@ -1,6 +1,8 @@
+import django
+
 from django.db import connection
 
-from .models import KeyPair, Region, Order, User
+from .models import KeyPair, Region, Order, User, WithDBColumn
 
 is_initialized = False
 
@@ -76,3 +78,14 @@ def setup_data():
     ]:
         parent = parent and KeyPair.objects.filter(key=parent).first()
         KeyPair.objects.create(key=key, value=value, parent=parent)
+
+    parent = None
+    for i in range(10):
+        parent = WithDBColumn.objects.create(parent=parent)
+
+    if django.VERSION >= (5, 2):
+        from .models import Site, WithCompositePK
+
+        site = Site.objects.create(name="test_site")
+
+        WithCompositePK.objects.create(site=site, username="test_user")

tests/models.py

@@ -1,12 +1,15 @@
+import django
+
 from django.db.models import (
-    CASCADE,
-    Manager,
-    Model,
-    QuerySet,
     AutoField,
+    CASCADE,
     CharField,
     ForeignKey,
     IntegerField,
+    Manager,
+    Model,
+    QuerySet,
+    SlugField,
     TextField,
 )
 
@@ -68,3 +71,22 @@ class KeyPair(Model):
 
     class Meta:
         db_table = "keypair"
+
+
+class WithDBColumn(Model):
+    id = AutoField(db_column="uid", primary_key=True)
+    parent = ForeignKey("self", db_column="pid", null=True, on_delete=CASCADE)
+
+
+if django.VERSION >= (5, 2):
+    from django.db.models import CompositePrimaryKey
+
+
+    class Site(Model):
+        name = SlugField()
+
+
+    class WithCompositePK(Model):
+        pk = CompositePrimaryKey("site_id", "username")
+        site = ForeignKey(Site, on_delete=CASCADE)
+        username = CharField(max_length=32)

tests/test_cte.py

@@ -5,13 +5,14 @@
 from django.db.models.expressions import (
     Exists, ExpressionWrapper, F, OuterRef, Subquery,
 )
+from django.db.models.deletion import Collector
 from django.db.models.sql.constants import LOUTER
 from django.db.utils import OperationalError, ProgrammingError
 from django.test import TestCase
 
 from django_cte import CTE, with_cte
 
-from .models import Order, Region, User
+from .models import Order, Region, User, WithDBColumn
 
 int_field = IntegerField()
 text_field = TextField()
@@ -359,6 +360,11 @@ def test_update_with_subquery(self):
         strict=True,
     )
     def test_delete_cte_query(self):
+        # This test requires "fast" deletion. If this constraint is broken
+        # the models have been modified in an incompatible way and they
+        # should be adjusted until this assertion passes again.
+        self.assertTrue(Collector(None).can_fast_delete(Order))
+
         cte = CTE(
             Order.objects
             .values(region_parent=F("region__parent_id"))
@@ -874,3 +880,40 @@ def test_left_outer_join_invalid_innerjoin(self):
             {'name': 'sun', 'total': None},
             {'name': 'venus', 'total': None}
         ])
+
+    def test_fields_with_db_column(self):
+        cte = CTE.recursive(
+            lambda cte: WithDBColumn.objects.filter(id=10)
+                .union(cte.join(WithDBColumn, id=cte.col.parent_id))
+        )
+        qs = with_cte(cte, select=cte)
+        query = str(qs.query)
+        self.assertIn("uid", query)
+        self.assertIn("pid", query)
+        self.assertEqual(
+            [(i.id, i.parent_id) for i in qs],
+            [(i, (i - 1) if i > 1 else None) for i in range(10, 0, -1)]
+        )
+
+    @pytest.mark.skipif(django.VERSION < (5, 2), reason='Requires Django 5.2+')
+    def test_composite_primary_key(self):
+        from .models import Site, WithCompositePK
+
+        cte = CTE(WithCompositePK.objects.all())
+        qs = with_cte(cte, select=cte).values(test=cte.col.pk)
+
+        with pytest.raises(ValueError, match='reference column pairs'):
+            str(qs.query)
+
+        qs = with_cte(cte, select=cte).values(test1=cte.col.site, test2=cte.col.username)
+        query = str(qs.query)
+        self.assertIn("test1", query)
+        self.assertIn("site_id", query)
+        self.assertIn("test2", query)
+        self.assertIn("username", query)
+        self.assertEqual(list(qs), [
+            {
+                'test1': Site.objects.values_list("pk", flat=True).get(),
+                'test2': "test_user",
+            }
+        ])