diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 8bfb5fa..9a57c94 100644 --- a/.github/workflows/tests.yml +++ b/.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: diff --git a/django_cte/cte.py b/django_cte/cte.py index d20e777..e3ba70f 100644 --- a/django_cte/cte.py +++ b/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: diff --git a/django_cte/meta.py b/django_cte/meta.py index b1865be..1988140 100644 --- a/django_cte/meta.py +++ b/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): diff --git a/django_cte/raw.py b/django_cte/raw.py index 104572f..b9131b1 100644 --- a/django_cte/raw.py +++ b/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) diff --git a/pyproject.toml b/pyproject.toml index 61861e9..77d970a 100644 --- a/pyproject.toml +++ b/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}, +] diff --git a/tests/django_setup.py b/tests/django_setup.py index b5c3f77..5b4f173 100644 --- a/tests/django_setup.py +++ b/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") diff --git a/tests/models.py b/tests/models.py index f3c9e0c..0b0f428 100644 --- a/tests/models.py +++ b/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) diff --git a/tests/test_cte.py b/tests/test_cte.py index a854f78..e63b74e 100644 --- a/tests/test_cte.py +++ b/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", + } + ])