diff --git a/.claude/rules/plain-postgres.md b/.claude/rules/plain-postgres.md index 68b58103b1..576c7b02b6 100644 --- a/.claude/rules/plain-postgres.md +++ b/.claude/rules/plain-postgres.md @@ -46,9 +46,21 @@ class Article(postgres.Model): - **String forward-ref FKs** (`"self"`, `"OtherModel"`) keep a _value-type_ annotation — the checker can't resolve the string to a model: `parent: Foo | None = types.ForeignKeyField("self", on_delete=postgres.CASCADE, allow_null=True, default=None)`. + That annotation is a model instance, not a field, so `where()` traversal + (`Child.parent.name.equals(...)`) doesn't type-check through it — declare the + target above and pass the class when you want traversal typed. +- **Encrypted fields** are annotated `EncryptedField[T]` (imported from + `plain.postgres` alongside `Field`), not `Field[T]`. It's a `Field[T]` + subclass, so the constructor is typed identically, but it also carries the + `Never`-typed blocks that reject `Model.secret.equals(...)` at the call site. + A plain `Field[T]` hides them and the comparison only fails at runtime. - **JSON**: `JSONField`/`EncryptedJSONField` return `Any` from the stub (the runtime class isn't generic over its value shape), so the annotation is what - preserves typing: `Field[dict]` / `Field[dict[str, Any]]`. + preserves typing: `Field[dict]` / `EncryptedField[dict[str, Any]]`. Never + annotate a field `Field[Any]` — `Any` satisfies the model-valued `__get__` + overload, so class access types as `type[Any]` and the whole condition + surface disappears silently. Use the concrete shape, or `Field[object]` when + the column really does hold arbitrary JSON. - **Custom querysets**: declare `query: ClassVar[MyQuerySet] = MyQuerySet()` (`ClassVar` so it isn't treated as a field). Default-queryset models declare nothing — `Model.query` is typed automatically. @@ -90,7 +102,11 @@ Run `uv run plain docs postgres` for full workflow details. Use `Model.query` to build querysets (e.g., `User.query.filter(is_active=True)`). +- `where()` takes typed conditions built off fields (`User.query.where(User.role.equals("admin"))`) instead of `filter()`'s string kwargs, so a typo or wrong value type is caught at the call site. +- **Conditions on a relation go through its key**: `Post.query.where(Post.author.id.equals(author.id))`, `.id.is_in([...])`, `.id.is_null()` — the typed spelling of `filter(author=author)`, same SQL. `Post.author.equals(author)` raises `AttributeError`: `Post.author` is `type[Author]` to the checker (which is what makes `Post.author.email.equals(...)` work), so it offers the related model's fields, not conditions. Traversal only _starts_ from a forward FK; a many-to-many is traversable as a later hop (`WidgetTag.widget.tags.name`), but `Widget.tags` and reverse accessors are not entry points — use `filter(tags__name=...)` there. +- Encrypted fields can't be looked up at all: `get_or_create(secret=...)` raises — put the value in `defaults=`. - Use `select_related()` for FK access in loops, `prefetch_related()` for reverse/M2N +- A foreign key with no value raises `RelatedObjectDoesNotExist`; catch it as `Related.DoesNotExist` (it subclasses that and `AttributeError`) — `Model.fk.RelatedObjectDoesNotExist` is a type error now that class access types as the related model. - A foreign key returns a partial related object: `obj.author` and `obj.author.id` are query-free; other fields load on first access. There is no `obj.author_id` — use `obj.author.id` - Use `.annotate(Count(...))` instead of calling `.count()` per row - Fetch all data in the view — templates should never trigger queries diff --git a/plain-cache/plain/cache/models.py b/plain-cache/plain/cache/models.py index 0767f42d2d..c02ae0cbd3 100644 --- a/plain-cache/plain/cache/models.py +++ b/plain-cache/plain/cache/models.py @@ -1,7 +1,7 @@ from __future__ import annotations from datetime import datetime -from typing import Any, ClassVar, Self +from typing import ClassVar, Self from plain.postgres import Field, types from plain.runtime import settings @@ -38,7 +38,13 @@ def forever(self) -> Self: @postgres.register_model class CachedItem(postgres.Model): key: Field[str] = types.TextField(max_length=255) - value: Field[Any] = types.JSONField(required=False, allow_null=True, default=None) + # `object`, not `Any`: the cache stores any JSON-serializable value, and + # `Field[Any]` would make `Any` satisfy the model-valued `__get__` overload, + # so `CachedItem.value` would type as `type[Any]` and lose the field surface + # entirely. + value: Field[object] = types.JSONField( + required=False, allow_null=True, default=None + ) expires_at: Field[datetime | None] = types.DateTimeField( required=False, allow_null=True, default=None ) diff --git a/plain-oauth/plain/oauth/models.py b/plain-oauth/plain/oauth/models.py index 643f16f435..71224d3e71 100644 --- a/plain-oauth/plain/oauth/models.py +++ b/plain-oauth/plain/oauth/models.py @@ -6,7 +6,7 @@ import psycopg from app.users.models import User from plain.exceptions import ValidationError -from plain.postgres import Field, transaction, types +from plain.postgres import EncryptedField, Field, transaction, types from plain.utils import timezone from plain import postgres @@ -36,8 +36,8 @@ class OAuthConnection(postgres.Model): provider_user_id: Field[str] = types.TextField(max_length=100) # Token data - access_token: Field[str] = types.EncryptedTextField(max_length=2000) - refresh_token: Field[str] = types.EncryptedTextField( + access_token: EncryptedField[str] = types.EncryptedTextField(max_length=2000) + refresh_token: EncryptedField[str] = types.EncryptedTextField( max_length=2000, required=False, default="" ) access_token_expires_at: Field[datetime | None] = types.DateTimeField( diff --git a/plain-postgres/plain/postgres/README.md b/plain-postgres/plain/postgres/README.md index 31f233f595..a15aa7f519 100644 --- a/plain-postgres/plain/postgres/README.md +++ b/plain-postgres/plain/postgres/README.md @@ -198,6 +198,63 @@ first_10_users = User.query.all()[:10] For more advanced querying options, see the [`QuerySet`](./query.py#QuerySet) class. +### Typed conditions with where() + +`where()` is a typed alternative to `filter()`. Instead of string keyword lookups, you build each condition from a field, so a type checker catches a misspelled field or a wrong value type at the call site: + +```python +from plain import postgres +from plain.postgres import Field, types + + +@postgres.register_model +class User(postgres.Model): + email: Field[str] = types.EmailField() + role: Field[str] = types.TextField(max_length=20) + age: Field[int | None] = types.IntegerField(allow_null=True, default=None) + + +# Each argument is a condition; multiple arguments are ANDed together. +admins = User.query.where( + User.role.equals("admin"), + User.age.gte(18), +) +``` + +Every field exposes `equals`, `not_equal`, `gt`, `gte`, `lt`, `lte`, `is_null`, and `is_in`. Text fields add `contains`, `icontains`, `startswith`, and `endswith`. Each returns a `Q`, so you can combine them with `|` and `&` or negate with `~`: + +```python +# Membership, negation, and OR +User.query.where(User.role.is_in(["admin", "staff"])) +User.query.where(~User.role.equals("guest")) +User.query.where(User.email.endswith("@example.com") | User.role.equals("admin")) +``` + +Conditions traverse foreign keys — accessing a field through a relation builds the joined lookup: + +```python +# Q(author__email="a@example.com") +Post.query.where(Post.author.email.equals("a@example.com")) +``` + +A relation is a path to traverse, not a field, so it carries no conditions of its own. To match on the relation itself, traverse to the key it points at — that's the typed spelling of `filter(author=author)`, and it compiles to the same SQL: + +```python +Post.query.where(Post.author.id.equals(author.id)) +Post.query.where(Post.author.id.is_in([a.id for a in authors])) +Post.query.where(Post.author.id.is_null()) # nullable relation +``` + +`Post.author.equals(author)` raises `AttributeError` naming this spelling (an `AttributeError`, so `hasattr` and `getattr(..., default)` keep behaving). It isn't an oversight: to the type checker `Post.author` is `type[Author]`, which is what makes `Post.author.email.equals(...)` type-check, and a condition method there would be a runtime method the checker rejects. + +Traversal starts from a **forward foreign key**. Once inside one, every relation you pass through is another hop, many-to-many included — `WidgetTag.widget.tags.name.equals("metal")` builds `Q(widget__tags__name="metal")` — and the same rule applies to the relation itself: `WidgetTag.widget.tags.equals(tag)` points you at `WidgetTag.widget.tags.id.equals(tag.id)`. + +A class-level many-to-many (`Widget.tags`) is _not_ an entry point: it has no traversal wiring, and it is typed `ManyToManyManager[Tag]`, so it could not be typed as one either. Use the string path there — `Widget.query.filter(tags__name="metal")`. Reverse relations aren't traversable for the same reason (a reverse accessor is a `ClassVar`, so there is nothing for the related model to offer the checker), and the error says so. + +A traversed field _is_ the related field, carrying the relation path as its name — so it offers exactly the conditions that field offers, including an encrypted field's refusals. + +[Encrypted fields](#encrypted-fields) reject value comparisons because their ciphertext is non-deterministic — only `is_null()` is available, and any other condition method (`equals`, `is_in`, …) raises `TypeError`. + ### Custom QuerySets You can customize [`QuerySet`](./query.py#QuerySet) classes to provide specialized query methods. Define a custom QuerySet and assign it to your model's `query` attribute as a `ClassVar` (so it isn't treated as a constructor field): @@ -991,28 +1048,44 @@ This is **not** for passwords or tokens you issue — those should be hashed (on ```python from plain import postgres -from plain.postgres import Field, types +from plain.postgres import EncryptedField, Field, types @postgres.register_model class Integration(postgres.Model): name: Field[str] = types.TextField(max_length=100) - api_key: Field[str] = types.EncryptedTextField(max_length=200) - credentials: Field[dict | None] = types.EncryptedJSONField( + api_key: EncryptedField[str] = types.EncryptedTextField(max_length=200) + credentials: EncryptedField[dict | None] = types.EncryptedJSONField( required=False, allow_null=True, default=None ) ``` +Annotate encrypted fields `EncryptedField[T]`, not `Field[T]`. The annotation is +what the type checker reads, and `EncryptedField[T]` is the `Field[T]` subclass +that declares the blocked conditions — with a plain `Field[T]`, +`Integration.api_key.equals("x")` type-checks its way to a runtime `TypeError` +instead of being rejected at the call site. It types the constructor exactly as +`Field[T]` does. + Values are encrypted using Fernet (AES-128-CBC + HMAC-SHA256) with a key derived from `SECRET_KEY`. The `cryptography` package is required — install it with `pip install cryptography`. **Available fields:** +- `EncryptedField[T]` — the annotation type; also the shared base the two fields below derive from. - `EncryptedTextField` — encrypts text, stored as `text` in the database regardless of `max_length` (ciphertext is longer than plaintext). `max_length` is enforced on the plaintext value during validation. - `EncryptedJSONField` — serializes to JSON, encrypts, and stores as `text`. Supports custom `encoder` and `decoder` parameters (same as `JSONField`). **Limitations:** -- **No lookups** — encrypted values are non-deterministic (same plaintext produces different ciphertext each time), so filtering on encrypted fields doesn't work. Only `isnull` lookups are supported. +- **No lookups** — encrypted values are non-deterministic (same plaintext produces different ciphertext each time), so filtering on encrypted fields doesn't work. Only `isnull` lookups are supported. Comparing against a value raises `TypeError` rather than silently matching nothing — both `filter(api_key="x")` and the typed [condition methods](#typed-conditions-with-where) (`equals`, `contains`, …), which are also rejected at the call site when the field is annotated `EncryptedField[T]`. `filter(api_key=None)` still rewrites to `IS NULL`. +- **`get_or_create()` must not look up an encrypted field.** `get_or_create(api_key="k")` raises, and the error says to move the value into `defaults=`. This is a deliberate break: it previously "worked" by creating a new row on every call, because the lookup could never match existing ciphertext. An encrypted value can be written, just not looked up: + + ```python + Integration.query.get_or_create(name="acme", defaults={"api_key": "k"}) + ``` + + The same applies to `update_or_create()`, and to an expression right-hand side like `filter(api_key=F("name"))` — the column is still ciphertext. + - **No indexes or constraints** — encrypted fields cannot be used in indexes or unique constraints. Preflight checks will catch this. - **Only `default=""`** — on `EncryptedTextField` (paired with `required=False`), the empty string is stored as plaintext `''`, so it's the one value expressible as a column `DEFAULT` (declare it to add the field to a populated table). Any other default would need ciphertext, which is non-deterministic. `EncryptedJSONField` has no persistent default at all — even `{}` serializes to text that would need ciphertext — so pair `allow_null=True` with `default=None`, which stores nothing and just marks the field optional in the constructor. @@ -1053,6 +1126,8 @@ book.author.name # one query — loads the rest of the row The first access to any non-key field loads the whole row in a single query. There is no separate `author_id` attribute — `book.author.id` is the foreign key value, and it is type-checked because `book.author` is an `Author`. In loops, use `select_related()` to load related rows up front and avoid a query per row. +A foreign key with no value raises `RelatedObjectDoesNotExist` on access. That attribute still lives on the descriptor at runtime, but class-level access is now typed as the related model (that is what makes `Book.author.name.equals(...)` work), so `Book.author.RelatedObjectDoesNotExist` is a type error. Catch it as `Author.DoesNotExist` — the exception subclasses both that and `AttributeError` — or as `AttributeError`. + The partial-instance shortcut is safe because Plain always creates a database foreign-key constraint, so the referenced row is guaranteed to exist. ### Constraints are checked immediately diff --git a/plain-postgres/plain/postgres/__init__.py b/plain-postgres/plain/postgres/__init__.py index dee44e8324..7b34ed355c 100644 --- a/plain-postgres/plain/postgres/__init__.py +++ b/plain-postgres/plain/postgres/__init__.py @@ -36,6 +36,7 @@ URLField, UUIDField, ) +from .fields.encrypted import EncryptedField from .fields.json import JSONField from .fields.timezones import TimeZoneField from .fields.related import ( @@ -70,6 +71,10 @@ "DecimalField", "DurationField", "EmailField", + # The typed descriptor base for encrypted fields, for annotating them: + # api_key: EncryptedField[str] = types.EncryptedTextField() + # It blocks the value-comparison conditions a plain Field[T] would allow. + "EncryptedField", "F", # The typed descriptor base, for annotating model fields: # name: Field[str] = types.TextField() diff --git a/plain-postgres/plain/postgres/agents/.claude/rules/plain-postgres.md b/plain-postgres/plain/postgres/agents/.claude/rules/plain-postgres.md index 68b58103b1..576c7b02b6 100644 --- a/plain-postgres/plain/postgres/agents/.claude/rules/plain-postgres.md +++ b/plain-postgres/plain/postgres/agents/.claude/rules/plain-postgres.md @@ -46,9 +46,21 @@ class Article(postgres.Model): - **String forward-ref FKs** (`"self"`, `"OtherModel"`) keep a _value-type_ annotation — the checker can't resolve the string to a model: `parent: Foo | None = types.ForeignKeyField("self", on_delete=postgres.CASCADE, allow_null=True, default=None)`. + That annotation is a model instance, not a field, so `where()` traversal + (`Child.parent.name.equals(...)`) doesn't type-check through it — declare the + target above and pass the class when you want traversal typed. +- **Encrypted fields** are annotated `EncryptedField[T]` (imported from + `plain.postgres` alongside `Field`), not `Field[T]`. It's a `Field[T]` + subclass, so the constructor is typed identically, but it also carries the + `Never`-typed blocks that reject `Model.secret.equals(...)` at the call site. + A plain `Field[T]` hides them and the comparison only fails at runtime. - **JSON**: `JSONField`/`EncryptedJSONField` return `Any` from the stub (the runtime class isn't generic over its value shape), so the annotation is what - preserves typing: `Field[dict]` / `Field[dict[str, Any]]`. + preserves typing: `Field[dict]` / `EncryptedField[dict[str, Any]]`. Never + annotate a field `Field[Any]` — `Any` satisfies the model-valued `__get__` + overload, so class access types as `type[Any]` and the whole condition + surface disappears silently. Use the concrete shape, or `Field[object]` when + the column really does hold arbitrary JSON. - **Custom querysets**: declare `query: ClassVar[MyQuerySet] = MyQuerySet()` (`ClassVar` so it isn't treated as a field). Default-queryset models declare nothing — `Model.query` is typed automatically. @@ -90,7 +102,11 @@ Run `uv run plain docs postgres` for full workflow details. Use `Model.query` to build querysets (e.g., `User.query.filter(is_active=True)`). +- `where()` takes typed conditions built off fields (`User.query.where(User.role.equals("admin"))`) instead of `filter()`'s string kwargs, so a typo or wrong value type is caught at the call site. +- **Conditions on a relation go through its key**: `Post.query.where(Post.author.id.equals(author.id))`, `.id.is_in([...])`, `.id.is_null()` — the typed spelling of `filter(author=author)`, same SQL. `Post.author.equals(author)` raises `AttributeError`: `Post.author` is `type[Author]` to the checker (which is what makes `Post.author.email.equals(...)` work), so it offers the related model's fields, not conditions. Traversal only _starts_ from a forward FK; a many-to-many is traversable as a later hop (`WidgetTag.widget.tags.name`), but `Widget.tags` and reverse accessors are not entry points — use `filter(tags__name=...)` there. +- Encrypted fields can't be looked up at all: `get_or_create(secret=...)` raises — put the value in `defaults=`. - Use `select_related()` for FK access in loops, `prefetch_related()` for reverse/M2N +- A foreign key with no value raises `RelatedObjectDoesNotExist`; catch it as `Related.DoesNotExist` (it subclasses that and `AttributeError`) — `Model.fk.RelatedObjectDoesNotExist` is a type error now that class access types as the related model. - A foreign key returns a partial related object: `obj.author` and `obj.author.id` are query-free; other fields load on first access. There is no `obj.author_id` — use `obj.author.id` - Use `.annotate(Count(...))` instead of calling `.count()` per row - Fetch all data in the view — templates should never trigger queries diff --git a/plain-postgres/plain/postgres/fields/base.py b/plain-postgres/plain/postgres/fields/base.py index 7dc58db0dc..0ae18aed90 100644 --- a/plain-postgres/plain/postgres/fields/base.py +++ b/plain-postgres/plain/postgres/fields/base.py @@ -3,7 +3,7 @@ import collections.abc import copy import enum -from collections.abc import Callable, Sequence +from collections.abc import Callable, Iterable, Sequence from functools import cached_property from typing import ( TYPE_CHECKING, @@ -16,7 +16,7 @@ from plain.postgres.constants import LOOKUP_SEP from plain.postgres.dialect import quote_name from plain.postgres.enums import ChoicesMeta -from plain.postgres.query_utils import RegisterLookupMixin +from plain.postgres.query_utils import Q, RegisterLookupMixin from plain.preflight import PreflightResult from plain.utils.datastructures import DictWrapper from plain.utils.functional import Promise @@ -97,6 +97,10 @@ def _empty(of_cls: type) -> Empty: return new +# Ordering conditions: the ones a None operand is meaningless for. +_ORDERING_SUFFIXES = frozenset({"gt", "gte", "lt", "lte"}) + + class Field[T](RegisterLookupMixin): """Base class for all field types""" @@ -143,6 +147,8 @@ def __str__(self) -> str: Return "package_label.model_label.field_name" for fields attached to models. """ + if self.is_lookup_reference: + return f"{self.name} (lookup reference)" if not hasattr(self, "model"): return super().__str__() model = self.model @@ -152,10 +158,137 @@ def __repr__(self) -> str: """Display the module, class, and name of the field.""" path = f"{self.__class__.__module__}.{self.__class__.__qualname__}" name = getattr(self, "name", "") + if self.is_lookup_reference: + return f"<{path} lookup reference: {name}>" if name: return f"<{path}: {name}>" return f"<{path}>" + # Typed query conditions. Available on every field; subclasses extend + # with type-specific lookups (comparison on numeric, string ops on text). + # The names are listed once in CONDITION_METHODS below -- anything that + # needs the set (traversal advice, tests) imports it rather than retyping. + def equals(self, value: T) -> Q: + return self._build_q("equals", "", value) + + def not_equal(self, value: T) -> Q: + return ~self._build_q("not_equal", "", value) + + def gt(self, value: T) -> Q: + return self._build_q("gt", "gt", value) + + def gte(self, value: T) -> Q: + return self._build_q("gte", "gte", value) + + def lt(self, value: T) -> Q: + return self._build_q("lt", "lt", value) + + def lte(self, value: T) -> Q: + return self._build_q("lte", "lte", value) + + def is_null(self, value: bool = True) -> Q: + return self._build_q("is_null", "isnull", value) + + def is_in(self, values: Iterable[T]) -> Q: + return self._build_q("is_in", "in", values) + + # Pattern conditions. They live on Field, like every other condition, so + # they survive the `Field[T]` annotation models carry -- a field's declared + # type is `Field[str]`, not `TextField[str]`, so the checker only ever sees + # what `Field` offers. The `self` annotation is the restriction: a + # `Field[int]` rejects `.startswith(...)` statically, and `_build_q` + # rejects it at runtime for anything that doesn't register the lookup. + # + # (`JSONField` does register a jsonb `contains`, so `Field[dict].contains` + # is rejected by the `self` type rather than at runtime -- the same + # type-first guard the encrypted fields use.) + def contains(self: Field[str] | Field[str | None], value: str) -> Q: + return self._build_q("contains", "contains", value) + + def icontains(self: Field[str] | Field[str | None], value: str) -> Q: + return self._build_q("icontains", "icontains", value) + + def startswith(self: Field[str] | Field[str | None], value: str) -> Q: + return self._build_q("startswith", "startswith", value) + + def endswith(self: Field[str] | Field[str | None], value: str) -> Q: + return self._build_q("endswith", "endswith", value) + + def _build_q(self, method: str, suffix: str, value: Any) -> Q: + """Build a Q from a lookup suffix + value. Uses Q's positional-tuple + constructor to bypass its reserved `_connector`/`_negated` kwargs that + confuse the type checker on `**{name: value}` expansion. + + `method` is the condition method's own name, used only for error + messages -- it's what the caller wrote, so it's what an error should + name. + """ + assert self.name, ( + "Field name must be set before building a query condition; " + "the field must be attached to a model." + ) + if value is None and suffix in _ORDERING_SUFFIXES: + # On a nullable field `T` includes None, so `age.gte(None)` gets + # past the type checker -- there is no way to subtract None from a + # TypeVar, so `self: Field[X | None], value: X` still solves X as + # `int | None` under both ty and pyright. Refuse here instead, at + # the call site, rather than letting it reach the compiler as a + # "Cannot use None as a query value" ValueError with no field in it. + raise TypeError( + f"{type(self).__name__} {self.name!r}: .{method}() has no " + f"meaning for None -- a SQL comparison against NULL is never " + f"true. Use .is_null() instead." + ) + if suffix == "in" and isinstance(value, str | bytes): + # `Iterable[T]` is satisfied by `str` when T is `str`, and `str` is + # a `Sequence[str]`, so there is no way to exclude it statically. + # Left alone it iterates characters and silently matches the wrong + # rows. + raise TypeError( + f"{type(self).__name__} {self.name!r}: .is_in() takes a " + f"collection of values, not a single {type(value).__name__}. " + f"Pass a list -- .is_in([{value!r}])." + ) + if suffix and not self.get_lookup(suffix): + # The type checker rejects most of these already (a `Field[int]` + # has no `.startswith`); this catches what it can't see. + raise TypeError( + f"{type(self).__name__} {self.name!r} does not support " + f".{method}() -- no {suffix!r} lookup is registered for it." + ) + name = f"{self.name}{LOOKUP_SEP}{suffix}" if suffix else self.name + return Q((name, value)) + + def with_lookup_prefix(self, prefix: str) -> Self: + """Return a detached copy of this field whose name carries `prefix`. + + This is all where() traversal needs: `Child.parent.name` hands back the + related model's own `name` field renamed to `parent__name`, so the + field's own condition methods build `Q(parent__name=...)`. Nothing has + to re-implement or rewrite the field's surface, which is why a + traversed field offers exactly what direct access offers -- including + an encrypted field's blocks, whose error message names the full path. + + The copy is genuinely detached: it keeps only what building a Q needs + (its class, for the lookup registry, and its name). The attachment + state a real field carries is dropped, so it can't pass itself off as + a column on the related model -- `str()` would otherwise report + `examples.DeleteParent.parent__name`, and `__reduce__` would try to + look up an attribute that doesn't exist. + """ + prefixed = copy.copy(self) + for attached in ("model", "column", "cached_col"): + prefixed.__dict__.pop(attached, None) + prefixed.name = f"{prefix}{LOOKUP_SEP}{self.name}" + prefixed.__dict__["_is_lookup_reference"] = True + return prefixed + + @property + def is_lookup_reference(self) -> bool: + """True for a field handed back by `with_lookup_prefix` -- a reference + to a column reached through a relation, not a column on a model.""" + return bool(self.__dict__.get("_is_lookup_reference")) + def preflight(self, **kwargs: Any) -> list[PreflightResult]: return [*self._check_field_name()] @@ -393,6 +526,19 @@ def contribute_to_class(self, cls: type[Model], name: str) -> None: setattr(cls, self.name, self) # Descriptor protocol implementation + # + # The first overload is class access on a *model-valued* field -- a foreign + # key, nullable or not. It yields `type[T]` rather than the descriptor so + # the related model's own typed field surface is reachable for where() + # traversal (`Child.parent.name.equals(...)`), matching the traversal + # `ForwardForeignKeyDescriptor.__getattr__` serves at runtime. It comes + # first so it wins over the plain `Self` overload for FK fields; a + # non-model T never matches it. + @overload + def __get__[M: Model]( + self: Field[M] | Field[M | None], instance: None, owner: type[Model] + ) -> type[M]: ... + @overload def __get__(self, instance: None, owner: type[Model]) -> Self: ... @@ -540,6 +686,24 @@ def value_from_object(self, obj: Model) -> T | None: return getattr(obj, self.name) +# The condition methods `Field` exposes, named once. Anything that needs the +# set rather than the methods themselves -- the relation-traversal advice in +# related_typed.py, the tests that sweep the surface -- imports from here +# instead of keeping its own copy in sync. +STRING_CONDITION_METHODS = ("contains", "icontains", "startswith", "endswith") +CONDITION_METHODS = ( + "equals", + "not_equal", + "gt", + "gte", + "lt", + "lte", + "is_null", + "is_in", + *STRING_CONDITION_METHODS, +) + + def validate_none_only_default( field: Field[Any], default: Any, *, allow_null: bool ) -> None: diff --git a/plain-postgres/plain/postgres/fields/encrypted.py b/plain-postgres/plain/postgres/fields/encrypted.py index c9924d4dcc..8a9db50eeb 100644 --- a/plain-postgres/plain/postgres/fields/encrypted.py +++ b/plain-postgres/plain/postgres/fields/encrypted.py @@ -3,7 +3,7 @@ import base64 import json from functools import cache -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Literal, Never, overload try: from cryptography.fernet import Fernet, InvalidToken, MultiFernet @@ -22,7 +22,7 @@ from plain import preflight -from .base import NOT_PROVIDED, validate_none_only_default +from .base import NOT_PROVIDED, Field, validate_none_only_default from .json import JSONField from .text import TextField @@ -31,9 +31,11 @@ from plain.postgres.connection import DatabaseConnection from plain.postgres.lookups import Lookup, Transform + from plain.postgres.query_utils import Q from plain.preflight.results import PreflightResult __all__ = [ + "EncryptedField", "EncryptedJSONField", "EncryptedTextField", ] @@ -108,43 +110,186 @@ def _decrypt(value: str) -> str: ) -class EncryptedFieldMixin: - """Shared behavior for all encrypted fields. +# Shared tail explaining why encrypted fields reject value comparisons. Kept +# free of advice, because the right advice differs by caller -- see the two +# refusal sites below. +_NON_DETERMINISTIC_EXPLANATION = ( + "encrypting the same plaintext twice produces different ciphertext, so an " + "equality comparison on the column can never match" +) + + +class _EncryptedExact(Exact): + """An exact lookup that rejects right-hand values ciphertext can't match. + + None passes through so the ORM's exact-None → isnull rewrite in + `build_lookup` still works, and so does any other value the field stores + deterministically (for text, the empty string, which is stored as + plaintext ''). Anything else could only ever match nothing, so it raises + instead of silently returning no rows. + """ + + def __init__(self, lhs: Any, rhs: Any) -> None: + # lhs.output_field is the encrypted field itself (the lookups.py idiom), + # so it decides which values have a deterministic stored form. + field = lhs.output_field + if not field.matches_deterministically(rhs): + # Its own sentence, not _lookup_unsupported_message's: `filter()` + # *is* supported here, just not against an arbitrary value. + raise TypeError( + f"Encrypted field {field.name!r} cannot be matched against " + f"this value: {_NON_DETERMINISTIC_EXPLANATION}. " + f"{field.matchable_values_hint()} If this came from " + f"get_or_create()/update_or_create(), move {field.name!r} into " + f"defaults= -- it can be written, just not looked up." + ) + super().__init__(lhs, rhs) + + +class EncryptedField[T](Field[T]): + """Shared base for all encrypted fields, and the type to annotate them with. Owns the lookup surface (isnull and exact only — ciphertext is non-deterministic) and the preflight that blocks indexes and unique - constraints. + constraints. Also blocks the typed-query comparison methods. + + Annotate encrypted model fields with this rather than the plain ``Field``: - Must be used with Field as a co-base class. + api_key: EncryptedField[str] = types.EncryptedTextField(max_length=200) + + The annotation is what the type checker sees, so a plain ``Field[str]`` + would hide the ``Never``-typed blocks below and let + ``Model.api_key.equals("x")`` type-check on its way to a runtime + ``TypeError``. It is a ``Field[T]`` subclass, so the synthesized + constructor types the field exactly as ``Field[T]`` would. + + Concrete encrypted fields mix this with the field they specialize + (``TextField``, ``JSONField``), which supplies the column behavior. """ - # Type hints for attributes provided by Field (the required co-base class) - name: str - model: Any + def __init__(self, **kwargs: Any) -> None: + # Present only for the type checker. ty resolves `super().__init__` in + # EncryptedJSONField through this class and lands on `Field.__init__`, + # which takes no arguments, so without a signature here it rejects the + # kwargs the concrete field forwards. At runtime this is a plain + # cooperative passthrough and the MRO would reach the concrete field + # either way. + super().__init__(**kwargs) # The complete lookup surface, replacing the base field's registry. # isnull is obviously needed. exact is required so that `filter(field=None)` # works — the ORM resolves "exact" first and then rewrites None to isnull. - # Exact lookups on non-None values will silently return no results (since - # ciphertext is non-deterministic), but blocking exact entirely would break - # the None/isnull path. The base classes are named directly — inheriting - # the concrete field's registrations would leak specialized lookups like - # JSONField's JSONExact, which compares against the jsonb 'null' literal - # and defeats the None→isnull rewrite. get_lookup()/get_transform() and - # registry consumers (e.g. unsupported-lookup error suggestions) all - # resolve through this one dict. A classmethod so both class-level and - # instance-level callers work. + # _EncryptedExact rejects non-None right-hand values, so the silent-no-rows + # behavior on `filter(field='something')` is blocked too. The base classes + # are named directly — inheriting the concrete field's registrations would + # leak specialized lookups like JSONField's JSONExact, which compares + # against the jsonb 'null' literal and defeats the None→isnull rewrite. + # get_lookup()/get_transform() and registry consumers (e.g. + # unsupported-lookup error suggestions) all resolve through this one dict. + # A classmethod so both class-level and instance-level callers work. @classmethod def get_lookups(cls) -> dict[str, type[Lookup | Transform]]: - return {"exact": Exact, "isnull": IsNull} + return {"exact": _EncryptedExact, "isnull": IsNull} def get_transform(self, name: str) -> Callable[..., Transform] | None: # JSONField's get_transform falls back to KeyTransformFactory for any # name — key transforms would operate on ciphertext, so block them. return None + def _build_q(self, method: str, suffix: str, value: Any) -> Q: + """The one runtime guard. Every condition method on `Field` funnels + through here, so blocking the ones that compare ciphertext takes a + single override -- including conditions that don't exist yet. + + What survives is exactly what the kwarg path allows, so + `field.equals(None)` and `filter(field=None)` can't disagree: + `isnull`, and equality against a value the field stores + deterministically. + """ + if suffix == "isnull" or ( + suffix == "" and self.matches_deterministically(value) + ): + return super()._build_q(method, suffix, value) + raise TypeError(self._lookup_unsupported_message(method)) + + def matches_deterministically(self, value: Any) -> bool: + """Whether `value` has a stored form equality can actually match. + + Only NULL, in general: everything else becomes ciphertext, and + encrypting the same plaintext twice gives different bytes. + """ + return value is None + + def matchable_values_hint(self) -> str: + return ".is_null() is the only condition it supports." + + if TYPE_CHECKING: + # The static half of the same block. `Never` as the parameter type + # rejects every call site; the return is `Never` (not `Q`) to reflect + # that control never returns, and `Never` is assignable to `Q` so + # `where(field.equals(...))` still type-checks at the use site with the + # parameter error as the one that surfaces. + # + # Declarations only -- `_build_q` above is what raises. Keeping them + # here means the static block and the runtime block can't drift into + # disagreeing about *how* to refuse, only about which methods exist. + # equals/not_equal are narrowed rather than blocked: the value types + # below are exactly the ones `matches_deterministically` accepts, so + # the static surface and the runtime guard agree. Everything else is + # `Never`. + # + # The `self` restriction is what distinguishes them -- a string-valued + # encrypted column stores "" as plaintext, a JSON one doesn't -- and it + # has to live here rather than on EncryptedTextField, because models + # annotate the field `EncryptedField[T]` and that annotation is all the + # checker sees. + @overload + def equals( + self: EncryptedField[str] | EncryptedField[str | None], + value: Literal[""] | None, + ) -> Q: ... + @overload + def equals(self, value: None) -> Q: ... # ty: ignore[invalid-method-override] + + @overload + def not_equal( + self: EncryptedField[str] | EncryptedField[str | None], + value: Literal[""] | None, + ) -> Q: ... + @overload + def not_equal(self, value: None) -> Q: ... # ty: ignore[invalid-method-override] + + def gt(self, value: Never) -> Never: ... # ty: ignore[invalid-method-override] + + def gte(self, value: Never) -> Never: ... # ty: ignore[invalid-method-override] + + def lt(self, value: Never) -> Never: ... # ty: ignore[invalid-method-override] + + def lte(self, value: Never) -> Never: ... # ty: ignore[invalid-method-override] + + def is_in(self, values: Never) -> Never: ... # ty: ignore[invalid-method-override] + + def contains(self, value: Never) -> Never: ... # ty: ignore[invalid-method-override] + + def icontains(self, value: Never) -> Never: ... # ty: ignore[invalid-method-override] + + def startswith(self, value: Never) -> Never: ... # ty: ignore[invalid-method-override] + + def endswith(self, value: Never) -> Never: ... # ty: ignore[invalid-method-override] + + def _lookup_unsupported_message(self, method: str) -> str: + assert self.name, ( + "Encrypted field must be attached to a model before its typed-query " + "methods can produce a meaningful error message." + ) + return ( + f"Encrypted field {self.name!r} does not support .{method}() " + f"against this value: {_NON_DETERMINISTIC_EXPLANATION}. " + f"{self.matchable_values_hint()}" + ) + def preflight(self, **kwargs: Any) -> list[PreflightResult]: - errors: list[PreflightResult] = super().preflight(**kwargs) # ty: ignore[unresolved-attribute] + errors: list[PreflightResult] = super().preflight(**kwargs) errors.extend(self._check_encrypted_constraints()) return errors @@ -190,7 +335,7 @@ def _check_encrypted_constraints(self) -> list[PreflightResult]: return errors -class EncryptedTextField[T: (str, str | None) = str](EncryptedFieldMixin, TextField[T]): +class EncryptedTextField[T: (str, str | None) = str](EncryptedField[T], TextField[T]): """A TextField that encrypts its value before storing in the database. Values are encrypted using Fernet (AES-128-CBC + HMAC-SHA256) with a key @@ -226,6 +371,21 @@ def __init__( validators=validators, ) + def matches_deterministically(self, value: Any) -> bool: + # `_encrypt("")` returns "" -- the empty string is stored as plaintext, + # which is exactly what makes `default=""` expressible as a column + # DEFAULT. So equality against it is meaningful, and + # `exclude(token="")` (the documented "unset" check) keeps working. + return super().matches_deterministically(value) or ( + isinstance(value, str) and value == "" + ) + + def matchable_values_hint(self) -> str: + return ( + '.is_null() and equality against "" (stored as plaintext) are the ' + "only conditions it supports." + ) + def get_db_prep_value( self, value: Any, connection: DatabaseConnection, prepared: bool = False ) -> Any: @@ -242,7 +402,7 @@ def from_db_value( return _decrypt(value) -class EncryptedJSONField(EncryptedFieldMixin, JSONField): +class EncryptedJSONField(EncryptedField[Any], JSONField): """A JSONField that encrypts its serialized value before storing in the database. The JSON value is serialized to a string, encrypted, and stored as text. diff --git a/plain-postgres/plain/postgres/fields/related_descriptors.py b/plain-postgres/plain/postgres/fields/related_descriptors.py index 1c4e502e88..f716be7cda 100644 --- a/plain-postgres/plain/postgres/fields/related_descriptors.py +++ b/plain-postgres/plain/postgres/fields/related_descriptors.py @@ -58,38 +58,38 @@ class Child(Model): """ def __init__(self, field_with_rel: Any) -> None: - self.field = field_with_rel + self._field = field_with_rel @cached_property def RelatedObjectDoesNotExist(self) -> type: # The exception can't be created at initialization time since the - # related model might not be resolved yet; `self.field.model` might + # related model might not be resolved yet; `self._field.model` might # still be a string model reference. return type( "RelatedObjectDoesNotExist", - (self.field.remote_field.model.DoesNotExist, AttributeError), + (self._field.remote_field.model.DoesNotExist, AttributeError), { - "__module__": self.field.model.__module__, - "__qualname__": f"{self.field.model.__qualname__}.{self.field.name}.RelatedObjectDoesNotExist", + "__module__": self._field.model.__module__, + "__qualname__": f"{self._field.model.__qualname__}.{self._field.name}.RelatedObjectDoesNotExist", }, ) - def is_cached(self, instance: Any) -> bool: - return self.field.is_cached(instance) + def _is_cached(self, instance: Any) -> bool: + return self._field.is_cached(instance) - def get_queryset(self) -> QuerySet: - qs = self.field.remote_field.model._model_meta.base_queryset + def _get_queryset(self) -> QuerySet: + qs = self._field.remote_field.model._model_meta.base_queryset return qs.all() - def get_prefetch_queryset( + def _get_prefetch_queryset( self, instances: list[Any], queryset: QuerySet | None = None ) -> tuple[QuerySet, Any, Any, bool, str, bool]: if queryset is None: - queryset = self.get_queryset() + queryset = self._get_queryset() - rel_obj_attr = self.field.get_foreign_related_value - instance_attr = self.field.get_local_related_value - related_field = self.field.target_field + rel_obj_attr = self._field.get_foreign_related_value + instance_attr = self._field.get_local_related_value + related_field = self._field.target_field # A foreign key is single-column, so prefetch with a join-less IN query. query = { @@ -102,13 +102,11 @@ def get_prefetch_queryset( rel_obj_attr, instance_attr, True, - self.field.get_cache_name(), + self._field.get_cache_name(), False, ) - def __get__( - self, instance: Any | None, cls: type | None = None - ) -> ForwardForeignKeyDescriptor | Any | None: + def __get__(self, instance: Any | None, cls: type | None = None) -> Any: """ Get the related instance through the forward relation. @@ -117,6 +115,10 @@ def __get__( - ``self`` is the descriptor managing the ``parent`` attribute - ``instance`` is the ``child`` instance - ``cls`` is the ``Child`` class (we don't need it) + + Class-level access (``Child.parent``) returns the descriptor itself, + the ordinary convention. Traversal for typed where() is served by + ``__getattr__`` below rather than by handing back a proxy. """ if instance is None: return self @@ -124,31 +126,86 @@ def __get__( # The related object is cached on the model state -- by select_related, # prefetch, the reverse accessor, a prior access, or assignment. try: - rel_obj = self.field.get_cached_value(instance) + rel_obj = self._field.get_cached_value(instance) except KeyError: # _get_raw_value loads the foreign key column on demand if it was # deferred (.only()/.defer()), so we always see the real key here. - pk_value = self.field._get_raw_value(instance) + pk_value = self._field._get_raw_value(instance) rel_obj = None if pk_value is not None: - remote_model = self.field.remote_field.model - target_name = self.field.target_field.name + remote_model = self._field.remote_field.model + target_name = self._field.target_field.name assert target_name is not None # The database FK constraint guarantees the row exists, so build # a partial related instance with only its primary key loaded -- # no query. Accessing any other field triggers the full-row # deferred load. rel_obj = remote_model.from_db([target_name], [pk_value]) - self.field.set_cached_value(instance, rel_obj) + self._field.set_cached_value(instance, rel_obj) # Checked on every access, including a cached None: a non-nullable # foreign key with no value must raise consistently, not just once. - if rel_obj is None and not self.field.allow_null: + if rel_obj is None and not self._field.allow_null: raise self.RelatedObjectDoesNotExist( - f"{self.field.model.__name__} has no {self.field.name}." + f"{self._field.model.__name__} has no {self._field.name}." ) return rel_obj + def _build_typed_ref(self) -> Any: + """The traversal entry point for this relation. + + Built lazily, and cached by `__getattr__` rather than by + `cached_property`: this can raise `UnresolvedRelationError`, and an + AttributeError raised inside a descriptor's `__get__` makes Python + fall back to `__getattr__`, which would replace the useful message + with a bare "no attribute '_typed_ref'". + + The import is local because `related_typed` reaches `fields.related`, + which imports this module at load time. + """ + from plain.postgres.fields.related_typed import ( + RelatedFieldRef, + unresolved_relation_error, + ) + + try: + model = self._field.remote_field.model + except TypeError: + # `ForeignObjectRel.model` raises TypeError while the target is + # still a string, which would escape __getattr__ as a TypeError and + # make `hasattr` raise instead of returning False. + raise unresolved_relation_error( + self._field.name, str(self._field.remote_field.model_ref) + ) from None + + return RelatedFieldRef( + model=model, + prefix=self._field.name, + target_name=self._field.target_field.name, + ) + + def __getattr__(self, name: str) -> Any: + """Walk class-level attribute access into the related model, so typed + where() can build joined lookups: + + Child.parent.name.equals("x") → Q(parent__name="x") + + Only names this descriptor doesn't define reach here, and every + attribute it does define is ``_``-prefixed -- which model fields can + never be (``Meta`` skips ``_``-prefixed attributes when it collects + them), so a related field can't be shadowed by descriptor internals. + + Delegated to `RelatedFieldRef` so the first hop follows exactly the + same rule as every later one. + """ + if name.startswith("_"): + # Internals, and anything a field could never be named. + raise AttributeError(name) + ref = self.__dict__.get("_typed_ref") + if ref is None: + ref = self.__dict__["_typed_ref"] = self._build_typed_ref() + return getattr(ref, name) + def __set__(self, instance: Any, value: Any) -> None: """ Set the related object (or its raw key) through the forward relation. @@ -165,19 +222,19 @@ def __set__(self, instance: Any, value: Any) -> None: if isinstance(value, LazyObject): value = value if value else None - name = self.field.name + name = self._field.name assert name is not None - remote_field = self.field.remote_field + remote_field = self._field.remote_field if value is None: instance.__dict__[name] = None - self.field.set_cached_value(instance, None) + self._field.set_cached_value(instance, None) return if isinstance(value, remote_field.model): # A related model instance: store its key, cache the object. - instance.__dict__[name] = getattr(value, self.field.target_field.name) - self.field.set_cached_value(instance, value) + instance.__dict__[name] = getattr(value, self._field.target_field.name) + self._field.set_cached_value(instance, value) return if isinstance(value, Model | bool): @@ -185,38 +242,40 @@ def __set__(self, instance: Any, value: Any) -> None: # the key 0/1 via int) -- reject rather than store a bogus key. raise TypeError( f'Cannot assign "{value!r}": ' - f'"{instance.model_options.object_name}.{self.field.name}" must be a ' + f'"{instance.model_options.object_name}.{self._field.name}" must be a ' f'"{remote_field.model.model_options.object_name}" instance or a ' f"primary key value." ) # A bare related key value (e.g. child.parent = 5). - new_value = self.field.to_python(value) + new_value = self._field.to_python(value) # On an actual key change, drop the now-stale forward cache. # Re-storing the same key (e.g. by clean_fields) keeps the cache. - if instance.__dict__.get(name) != new_value and self.field.is_cached(instance): - self.field.delete_cached_value(instance) + if instance.__dict__.get(name) != new_value and self._field.is_cached(instance): + self._field.delete_cached_value(instance) instance.__dict__[name] = new_value def __delete__(self, instance: Any) -> None: """Delete the foreign key value, clearing any cached related object.""" try: - del instance.__dict__[self.field.name] + del instance.__dict__[self._field.name] except KeyError: raise AttributeError( f"{instance.__class__.__name__!r} object has no attribute " - f"{self.field.name!r}" + f"{self._field.name!r}" ) - if self.field.is_cached(instance): - self.field.delete_cached_value(instance) + if self._field.is_cached(instance): + self._field.delete_cached_value(instance) def __reduce__(self) -> tuple[Any, tuple[Any, str]]: """ - Pickling should return the instance attached by self.field on the - model, not a new copy of that descriptor. Use getattr() to retrieve - the instance directly from the model. + Pickling should return the instance attached by self._field on the + model, not a new copy of that descriptor. + + Class access returns the descriptor, so ``getattr`` retrieves the + instance directly from the model. """ - return getattr, (self.field.model, self.field.name) + return getattr, (self._field.model, self._field.name) class ForwardManyToManyDescriptor: diff --git a/plain-postgres/plain/postgres/fields/related_managers.py b/plain-postgres/plain/postgres/fields/related_managers.py index 2533ab1f65..6c307bf417 100644 --- a/plain-postgres/plain/postgres/fields/related_managers.py +++ b/plain-postgres/plain/postgres/fields/related_managers.py @@ -139,7 +139,7 @@ def get_queryset(self) -> QS: queryset = self.model.query return cast(QS, self._apply_rel_filters(queryset)) - def get_prefetch_queryset( + def _get_prefetch_queryset( self, instances: Iterable[Model], queryset: QuerySet | None = None ) -> tuple[ QuerySet, Callable[[Model], Any], Callable[[Model], Any], bool, str, bool @@ -385,7 +385,7 @@ def get_queryset(self) -> QS: queryset = self.model.query return cast(QS, self._apply_rel_filters(queryset)) - def get_prefetch_queryset( + def _get_prefetch_queryset( self, instances: Iterable[Model], queryset: QuerySet | None = None ) -> tuple[ QuerySet, Callable[[Model], Any], Callable[[Model], Any], bool, str, bool diff --git a/plain-postgres/plain/postgres/fields/related_typed.py b/plain-postgres/plain/postgres/fields/related_typed.py new file mode 100644 index 0000000000..09e853f3f7 --- /dev/null +++ b/plain-postgres/plain/postgres/fields/related_typed.py @@ -0,0 +1,157 @@ +"""Typed FK traversal for the where() query API. + +When `Order.user` is a ForeignKey, accessing `.email` at the class level (as in +`where(Order.user.email.equals("x"))`) needs to produce `Q(user__email="x")` so +the existing SQL builder's join machinery resolves the right column. + +The whole mechanism is `Field.with_lookup_prefix`: walking into the related +model hands back that model's own field, renamed to carry the relation path, so +its own condition methods build the right keys. Nothing here re-implements or +rewrites a field's surface -- which is why a traversed field offers exactly +what direct access offers, down to an encrypted field's blocks. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from plain.postgres.constants import LOOKUP_SEP +from plain.postgres.exceptions import FieldDoesNotExist +from plain.postgres.fields.base import CONDITION_METHODS +from plain.postgres.fields.related import RelatedField +from plain.postgres.fields.reverse_descriptors import ( + ReverseForeignKey, + ReverseManyToMany, +) + +if TYPE_CHECKING: + from plain.postgres.base import Model + + +class UnresolvedRelationError(AttributeError): + """A traversal reached a relation whose target model isn't resolved yet. + + An AttributeError subclass so `hasattr` and `getattr(..., default)` keep + working, but a distinct type so the timing problem is greppable rather + than looking like a typo. + """ + + +class RelatedFieldRef: + """Class-level proxy that walks attribute access into the related model and + accumulates the lookup-path prefix as it goes. + + Chained traversal (`Order.user.profile.city`) builds nested + `RelatedFieldRef` instances until a concrete field is reached, which comes + back as a prefixed copy of that field. Every relation is a hop -- many-to- + many included, since `widget__tags__name` is as valid a lookup path as + `widget__author__name`. + + Names resolve through the related model's metadata (`get_forward_field`), + not attribute lookup, so a related field keeps resolving to the field. + + A relation is not itself a field, so it carries no condition methods. + `Child.parent.equals(obj)` is spelled `Child.parent.id.equals(obj.id)` -- + traversal to the key the relation targets, which compiles to the same + `parent__id=` lookup `filter(parent=obj)` produces. This can't be smoothed + over by adding the methods here: to the type checker `Child.parent` is + `type[Parent]` (see `Field.__get__`'s model-valued overloads), which is + what makes chained traversal type-check, and a runtime method the checker + rejects would be worse than no method at all. `__getattr__` raises an + AttributeError pointing at the right spelling instead. + """ + + def __init__(self, model: type[Model], prefix: str, target_name: str) -> None: + if isinstance(model, str): + raise unresolved_relation_error(prefix, model) + self._model = model + self._prefix = prefix + # The field on the related model that this relation targets -- the hop + # a condition on the relation has to go through. + self._target_name = target_name + + def __repr__(self) -> str: + return f"" + + def _is_reverse_relation(self, name: str) -> bool: + """Whether `name` names a reverse accessor on the related model. + + Reverse accessors live on the class as descriptors, not in the field + metadata, so both places are checked. + """ + try: + self._model._model_meta.get_reverse_relation(name) + except FieldDoesNotExist: + return isinstance( + getattr(self._model, name, None), + (ReverseForeignKey, ReverseManyToMany), + ) + return True + + @property + def _attribute_path(self) -> str: + """The prefix spelled the way it was written -- `widget.tags`, not + `widget__tags` -- so error messages can be pasted back into code.""" + return self._prefix.replace(LOOKUP_SEP, ".") + + def __getattr__(self, name: str) -> Any: + if name.startswith("_"): + # Avoid infinite recursion on internals and let pickling/hasattr + # checks fail cleanly. + raise AttributeError(name) + + try: + field = self._model._model_meta.get_forward_field(name) + except FieldDoesNotExist: + # The field lookup comes first so a related model that really does + # have a column named `equals` (or `contains`, …) still traverses + # to it. Every failure below is an AttributeError, not a TypeError, + # so `hasattr` and `getattr(..., default)` keep behaving. + if name in CONDITION_METHODS: + raise AttributeError( + f"{self._attribute_path}.{name}() is not available: " + f"{self._prefix!r} is a relation, not a field. Build the " + f"condition on the key it points at instead -- " + f"{self._attribute_path}.{self._target_name}.{name}(...), " + f"which compiles to the same SQL." + ) from None + if self._is_reverse_relation(name): + raise AttributeError( + f"{self._attribute_path}.{name} is a reverse relation, " + f"which the typed API cannot traverse: a reverse accessor " + f"is a ClassVar, so there is nothing for " + f"`{self._model.__name__}` to offer the type checker here. " + f"Use the string path instead -- " + f"filter({self._prefix}{LOOKUP_SEP}{name}{LOOKUP_SEP}...=...)." + ) from None + raise AttributeError( + f"{self._attribute_path}.{name} is not a traversable field or relation" + ) from None + + if isinstance(field, RelatedField): + # Any relation is another hop, foreign key or many-to-many alike -- + # `widget__tags__name` is as valid a lookup path as + # `widget__author__name`. Handing back the relation field itself + # would rename it to "widget__tags" and then let `.name` resolve to + # that string. + return RelatedFieldRef( + model=field.remote_field.model, + prefix=f"{self._prefix}{LOOKUP_SEP}{name}", + target_name=field.target_field.name, + ) + return field.with_lookup_prefix(self._prefix) + + +def unresolved_relation_error(prefix: str, target: str) -> UnresolvedRelationError: + """The error for a traversal that outran model registration. + + Relation targets are replaced with the resolved class when the model + registers, so a traversal evaluated at import time -- at module level, or + in a default argument -- can run before the registry is populated. + """ + return UnresolvedRelationError( + f"Cannot traverse {prefix!r}: its target model {target!r} hasn't been " + f"resolved yet. Relation targets are resolved when the model is " + f"registered, so this traversal is running too early -- move it inside " + f"the function or method that needs it." + ) diff --git a/plain-postgres/plain/postgres/query.py b/plain-postgres/plain/postgres/query.py index a1cca8ed96..d44ed5e4ca 100644 --- a/plain-postgres/plain/postgres/query.py +++ b/plain-postgres/plain/postgres/query.py @@ -1128,6 +1128,18 @@ def exclude(self, *args: Any, **kwargs: Any) -> Self: """ return self._filter_or_exclude(True, args, kwargs) + def where(self, *conditions: Q) -> Self: + """ + Return a new QuerySet narrowed by typed field conditions. + + Conditions are produced by field methods like `Model.field.equals(...)` + and combine with `|` and `&`. Unlike `filter()`, this accepts no + keyword arguments — every condition is a typed expression, so a + type checker can reject typos and value-type mismatches at the call + site. + """ + return self.filter(*conditions) + def _filter_or_exclude( self, negate: bool, args: tuple[Any, ...], kwargs: dict[str, Any] ) -> Self: @@ -1886,9 +1898,9 @@ def get_prefetcher( ) -> tuple[Any, Any, bool, Callable[[Model], bool]]: """ For the attribute 'through_attr' on the given instance, find - an object that has a get_prefetch_queryset(). + an object that has a _get_prefetch_queryset(). Return a 4 tuple containing: - (the object with get_prefetch_queryset (or None), + (the object with _get_prefetch_queryset (or None), the descriptor object representing this relationship (or None), a boolean that is False if the attribute was not found at all, a function that takes an instance and returns a boolean that is True if @@ -1911,16 +1923,16 @@ def has_to_attr_attribute(instance: Model) -> bool: attr_found = True if rel_obj_descriptor: # singly related object, descriptor object has the - # get_prefetch_queryset() method. - if hasattr(rel_obj_descriptor, "get_prefetch_queryset"): + # _get_prefetch_queryset() method. + if hasattr(rel_obj_descriptor, "_get_prefetch_queryset"): prefetcher = rel_obj_descriptor - is_fetched = rel_obj_descriptor.is_cached + is_fetched = rel_obj_descriptor._is_cached else: # descriptor doesn't support prefetching, so we go ahead and get # the attribute on the instance rather than the class to # support many related managers rel_obj = getattr(instance, through_attr) - if hasattr(rel_obj, "get_prefetch_queryset"): + if hasattr(rel_obj, "_get_prefetch_queryset"): prefetcher = rel_obj if through_attr != to_attr: # Special case cached_property instances because hasattr @@ -1954,7 +1966,7 @@ def prefetch_one_level( Return the prefetched objects along with any additional prefetches that must be done due to prefetch_related lookups found from default managers. """ - # prefetcher must have a method get_prefetch_queryset() which takes a list + # prefetcher must have a method _get_prefetch_queryset() which takes a list # of instances, and returns a tuple: # (queryset of instances of self.model that are related to passed in instances, @@ -1974,7 +1986,7 @@ def prefetch_one_level( single, cache_name, is_descriptor, - ) = prefetcher.get_prefetch_queryset(instances, lookup.get_current_queryset(level)) + ) = prefetcher._get_prefetch_queryset(instances, lookup.get_current_queryset(level)) # We have to handle the possibility that the QuerySet we just got back # contains some prefetch_related lookups. We don't want to trigger the # prefetch_related functionality by evaluating the query. Rather, we need diff --git a/plain-postgres/plain/postgres/types.pyi b/plain-postgres/plain/postgres/types.pyi index 81cee9f81a..93eb6290c1 100644 --- a/plain-postgres/plain/postgres/types.pyi +++ b/plain-postgres/plain/postgres/types.pyi @@ -9,7 +9,7 @@ typed *descriptor* (`XField[T]`), not the primitive `T`. Combined with email: Field[str] = types.EmailField() age: Field[int | None] = types.IntegerField(allow_null=True, default=None) - User.email # EmailField[str] — typed reference + User.email # EmailField[str] — typed reference, has .equals(), .contains(), ... user.email # str — the loaded value User.age # IntegerField[int | None] user.age # int | None @@ -514,16 +514,22 @@ def EncryptedJSONField( # Two overload families: # # 1. Class-argument FK (`to=SomeModel`) — T is inferred from the class. -# Returns `_ForeignKeyDescriptor[T, V]` whose `__get__` overloads keep -# class-access typed as the descriptor itself (matching runtime -# `ForwardForeignKeyDescriptor.__get__` which returns `self` when -# `instance is None`) and instance-access as the related instance -# (or `T | None` for nullable FKs). +# Returns `_ForeignKeyDescriptor[T, V]`, a `Field[V]`. `Field.__get__` +# does the double duty: class-access (`Child.parent`) yields `type[T]` so +# the related model's typed field surface (e.g. +# `Child.parent.name.equals(...)`) is visible for typed where() chaining; +# instance-access (`child.parent`) yields V (T or T | None for nullable +# FKs). Both survive the `Field[T]` annotation the model declares. # # 2. String-argument FK (`to="SomeModel"`, `to="self"`) — T can't be # inferred from the string, so the return type falls back to bare `T`. -# This requires an explicit LHS annotation (`parent: TreeNode | None = …`) -# but preserves typing for forward references and self-references. +# This requires an explicit *value-type* LHS annotation +# (`parent: TreeNode | None = …`) which preserves instance-access typing +# for forward references and self-references, but makes class access a +# `TreeNode` rather than a field: type-level FK traversal isn't available +# through string-arg FKs. The runtime `RelatedFieldRef` still resolves +# `Child.parent.name` regardless. Declare the model before the FK and pass +# the class when you want traversal to type-check. # # `__set__` accepts the related instance, None (via V), or a bare PK # value (int) — matching what `ForwardForeignKeyDescriptor` already @@ -536,11 +542,12 @@ def EncryptedJSONField( class _ForeignKeyDescriptor[T: Model, V](_Field[V]): # Subclasses Field[V] so an FK field is assignable to a `Field[V]` # annotation (e.g. `org: Field[Org] = types.ForeignKeyField(Org)`) under - # both ty and pyright. The FK-specific __get__/__set__ override the base. - @overload - def __get__(self, instance: None, owner: type) -> _ForeignKeyDescriptor[T, V]: ... - @overload - def __get__(self, instance: Model, owner: type) -> V: ... + # both ty and pyright. That annotation is also what models actually carry, + # so `__get__` is deliberately NOT overridden here -- `Field.__get__`'s + # model-valued overloads already give class access `type[T]` (traversal) + # and instance access `V`, and they keep working through the `Field[V]` + # annotation, which an override here would not. Only `__set__` is + # widened, to accept a bare PK alongside the instance. def __set__(self, instance: Model, value: V | int) -> None: ... # Class-argument FK overloads diff --git a/plain-postgres/tests/app/examples/migrations/0019_shadowtarget_shadowsource.py b/plain-postgres/tests/app/examples/migrations/0019_shadowtarget_shadowsource.py new file mode 100644 index 0000000000..b73b6ffc0a --- /dev/null +++ b/plain-postgres/tests/app/examples/migrations/0019_shadowtarget_shadowsource.py @@ -0,0 +1,34 @@ +# Generated by Plain 0.154.0 on 2026-07-23 15:32 + +from plain.postgres import migrations + +from plain import postgres + + +class Migration(migrations.Migration): + dependencies = (("examples", "0018_storageparametersexample"),) + + operations = ( + migrations.CreateModel( + name="ShadowTarget", + fields=[ + ("id", postgres.PrimaryKeyField()), + ("field", postgres.TextField(max_length=100)), + ("get_prefetch_queryset", postgres.TextField(max_length=100)), + ("get_queryset", postgres.TextField(max_length=100)), + ("is_cached", postgres.TextField(max_length=100)), + ], + ), + migrations.CreateModel( + name="ShadowSource", + fields=[ + ("id", postgres.PrimaryKeyField()), + ( + "ref", + postgres.ForeignKeyField( + on_delete=postgres.CASCADE, to="examples.shadowtarget" + ), + ), + ], + ), + ) diff --git a/plain-postgres/tests/app/examples/migrations/0020_stringconditionsexample.py b/plain-postgres/tests/app/examples/migrations/0020_stringconditionsexample.py new file mode 100644 index 0000000000..ecfba25b03 --- /dev/null +++ b/plain-postgres/tests/app/examples/migrations/0020_stringconditionsexample.py @@ -0,0 +1,21 @@ +# Generated by Plain 0.163.1 on 2026-09-19 02:31 + +from plain.postgres import migrations + +from plain import postgres + + +class Migration(migrations.Migration): + dependencies = (("examples", "0019_shadowtarget_shadowsource"),) + + operations = ( + migrations.CreateModel( + name="StringConditionsExample", + fields=[ + ("id", postgres.PrimaryKeyField()), + ("ip", postgres.GenericIPAddressField()), + ("label", postgres.TextField(max_length=50)), + ("token", postgres.RandomStringField(length=16)), + ], + ), + ) diff --git a/plain-postgres/tests/app/examples/models/__init__.py b/plain-postgres/tests/app/examples/models/__init__.py index 66e6b7039b..6fc95fa0d4 100644 --- a/plain-postgres/tests/app/examples/models/__init__.py +++ b/plain-postgres/tests/app/examples/models/__init__.py @@ -13,7 +13,9 @@ nullability, querysets, relationships, + shadowing, storage_parameters, + string_conditions, trees, unregistered, ) diff --git a/plain-postgres/tests/app/examples/models/encrypted.py b/plain-postgres/tests/app/examples/models/encrypted.py index 38571362df..66517470ee 100644 --- a/plain-postgres/tests/app/examples/models/encrypted.py +++ b/plain-postgres/tests/app/examples/models/encrypted.py @@ -1,6 +1,6 @@ from __future__ import annotations -from plain.postgres import Field, types +from plain.postgres import EncryptedField, Field, types from plain import postgres @@ -10,8 +10,8 @@ class SecretStore(postgres.Model): """Model for testing encrypted fields.""" name: Field[str] = types.TextField(max_length=100) - api_key: Field[str] = types.EncryptedTextField(max_length=200) - notes: Field[str] = types.EncryptedTextField(required=False, default="") - config: Field[dict | None] = types.EncryptedJSONField( + api_key: EncryptedField[str] = types.EncryptedTextField(max_length=200) + notes: EncryptedField[str] = types.EncryptedTextField(required=False, default="") + config: EncryptedField[dict | None] = types.EncryptedJSONField( required=False, allow_null=True, default=None ) diff --git a/plain-postgres/tests/app/examples/models/relationships.py b/plain-postgres/tests/app/examples/models/relationships.py index b6f1927ea1..9ea9b6574c 100644 --- a/plain-postgres/tests/app/examples/models/relationships.py +++ b/plain-postgres/tests/app/examples/models/relationships.py @@ -16,19 +16,11 @@ class Tag(postgres.Model): ) -@postgres.register_model -class WidgetTag(postgres.Model): - """Through model for Widget-Tag many-to-many relationship.""" - - widget: Widget = types.ForeignKeyField("Widget", on_delete=postgres.CASCADE) - tag: Field[Tag] = types.ForeignKeyField(Tag, on_delete=postgres.CASCADE) - - @postgres.register_model class Widget(postgres.Model): name: Field[str] = types.TextField(max_length=100) size: Field[str] = types.TextField(max_length=100) - tags: types.ManyToManyManager[Tag] = types.ManyToManyField(Tag, through=WidgetTag) + tags: types.ManyToManyManager[Tag] = types.ManyToManyField(Tag, through="WidgetTag") model_options = postgres.Options( constraints=[ @@ -37,3 +29,14 @@ class Widget(postgres.Model): ), ] ) + + +# Declared after Widget so the FK takes the model *class* rather than a string: +# only a class-argument FK gives the stub a resolvable `T`, which is what makes +# `WidgetTag.widget.name.equals(...)` type-check as well as run. +@postgres.register_model +class WidgetTag(postgres.Model): + """Through model for Widget-Tag many-to-many relationship.""" + + widget: Field[Widget] = types.ForeignKeyField(Widget, on_delete=postgres.CASCADE) + tag: Field[Tag] = types.ForeignKeyField(Tag, on_delete=postgres.CASCADE) diff --git a/plain-postgres/tests/app/examples/models/shadowing.py b/plain-postgres/tests/app/examples/models/shadowing.py new file mode 100644 index 0000000000..814871b84f --- /dev/null +++ b/plain-postgres/tests/app/examples/models/shadowing.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +from plain.postgres import Field, types + +from plain import postgres + + +@postgres.register_model +class ShadowTarget(postgres.Model): + """Related model whose field names collide with public attributes on + ForwardForeignKeyDescriptor. Traversal through the FK must resolve these + to the fields, not the descriptor's own attributes.""" + + field: Field[str] = types.TextField(max_length=100) + is_cached: Field[str] = types.TextField(max_length=100) + get_queryset: Field[str] = types.TextField(max_length=100) + get_prefetch_queryset: Field[str] = types.TextField(max_length=100) + + +@postgres.register_model +class ShadowSource(postgres.Model): + ref: Field[ShadowTarget] = types.ForeignKeyField( + ShadowTarget, on_delete=postgres.CASCADE + ) diff --git a/plain-postgres/tests/app/examples/models/string_conditions.py b/plain-postgres/tests/app/examples/models/string_conditions.py new file mode 100644 index 0000000000..ec92ea763f --- /dev/null +++ b/plain-postgres/tests/app/examples/models/string_conditions.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +from plain.postgres import Field, types + +from plain import postgres + + +@postgres.register_model +class StringConditionsExample(postgres.Model): + """String-valued fields that are *not* TextField subclasses. + + `Field` declares the pattern conditions for every string-valued field, so + these have to carry them at runtime too — `GenericIPAddressField` is a + DefaultableField and `RandomStringField` is a ColumnField, and neither + inherits TextField's implementations. + """ + + label: Field[str] = types.TextField(max_length=50) + ip: Field[str] = types.GenericIPAddressField() + token: Field[str] = types.RandomStringField(length=16) diff --git a/plain-postgres/tests/internal/test_typed_where_internals.py b/plain-postgres/tests/internal/test_typed_where_internals.py new file mode 100644 index 0000000000..c430f4204b --- /dev/null +++ b/plain-postgres/tests/internal/test_typed_where_internals.py @@ -0,0 +1,80 @@ +"""Internals of the typed where() surface. + +These reach past the documented API -- `CONDITION_METHODS`, `_model_meta`, +`Field.with_lookup_prefix` -- to pin the *mechanism* rather than the contract. +The contract lives in `tests/public/test_typed_where.py` and +`test_typed_where_fk.py`; if these fail and those don't, something shifted +under the hood and you get to decide whether it should have. +""" + +from __future__ import annotations + +import pytest +from app.examples.models.delete import ChildCascade, DeleteParent +from app.examples.models.encrypted import SecretStore +from plain.postgres.fields.base import CONDITION_METHODS + + +def test_traversal_hands_back_the_field_itself(): + """The traversed object is the related model's own field, renamed. That is + what makes its surface identical to direct access by construction, rather + than by a delegation list someone has to keep in sync.""" + direct = DeleteParent._model_meta.get_forward_field("name") + traversed = ChildCascade.parent.name + + assert type(traversed) is type(direct) + assert traversed.name == "parent__name" + assert direct.name == "name" # the original is untouched + + +@pytest.mark.parametrize("method", CONDITION_METHODS) +def test_every_condition_name_gets_the_relation_advice(method): + """Sweep the whole condition surface, so a method added later can't quietly + fall through to the generic "not a traversable field" message.""" + with pytest.raises(AttributeError) as excinfo: + getattr(ChildCascade.parent, method) + + message = str(excinfo.value) + assert "is a relation, not a field" in message + assert f"parent.id.{method}(...)" in message + + +class TestEncryptedFieldTraversalBlocked: + """An encrypted field's refusals travel with it, because traversal hands + back the field itself. The message names the full path, since the prefixed + copy carries it as its name.""" + + @pytest.fixture + def traversed(self): + # No model in the examples app has an FK to SecretStore, so prefix the + # field directly. This is exactly what RelatedFieldRef hands back. + return SecretStore._model_meta.get_forward_field("api_key").with_lookup_prefix( + "store" + ) + + @pytest.mark.parametrize("method", [m for m in CONDITION_METHODS if m != "is_null"]) + def test_traversed_condition_raises(self, traversed, method): + with pytest.raises( + TypeError, match=rf"store__api_key.*does not support \.{method}\(" + ): + getattr(traversed, method)("x") + + def test_traversed_is_null_still_works(self, traversed): + assert traversed.is_null().children == [("store__api_key__isnull", True)] + + +def test_traversal_before_the_target_resolves_says_so(): + """A relation's target is a string until the model registers. A traversal + that runs at import time can land here first, and the old code dereferenced + `str._model_meta` -- an AttributeError that `hasattr` then swallowed, so + the symptom was a missing attribute rather than a timing problem.""" + from plain.postgres.fields.related_typed import ( + RelatedFieldRef, + UnresolvedRelationError, + ) + + with pytest.raises(UnresolvedRelationError, match=r"'Tag' hasn't been resolved"): + RelatedFieldRef(model="Tag", prefix="tags", target_name="id") # ty: ignore[invalid-argument-type] + + # An AttributeError subclass, so the attribute protocol still holds. + assert issubclass(UnresolvedRelationError, AttributeError) diff --git a/plain-postgres/tests/public/test_encrypted_fields.py b/plain-postgres/tests/public/test_encrypted_fields.py index eb7b997a79..3dfdc82740 100644 --- a/plain-postgres/tests/public/test_encrypted_fields.py +++ b/plain-postgres/tests/public/test_encrypted_fields.py @@ -1,7 +1,10 @@ from __future__ import annotations +from typing import assert_type + import pytest from app.examples.models.encrypted import SecretStore +from plain.postgres import F from plain.postgres.exceptions import FieldError from plain.postgres.fields.encrypted import ( _ENCRYPTED_PREFIX, @@ -9,6 +12,7 @@ _encrypt, _get_fernet, ) +from plain.postgres.query_utils import Q class TestEncryptDecryptFunctions: @@ -153,6 +157,168 @@ def test_unsupported_lookup_raises_field_error(self): SecretStore.query.filter(config__has_key="token") +class TestTypedQueryMethodsBlocked: + """Encrypted fields must not expose typed-query comparison methods. + + The class-level overrides accept `Never`, so type checkers reject any + call site. The runtime also raises TypeError as a safety net for callers + that bypass type checking (e.g. dynamic code). + """ + + def test_equals_raises(self): + with pytest.raises(TypeError, match=r"api_key.*does not support \.equals\("): + SecretStore.api_key.equals("anything") # ty: ignore[no-matching-overload] + + def test_not_equal_raises(self): + with pytest.raises(TypeError, match=r"does not support \.not_equal\("): + SecretStore.api_key.not_equal("x") # ty: ignore[no-matching-overload] + + @pytest.mark.parametrize("method", ["gt", "gte", "lt", "lte"]) + def test_ordering_comparison_raises(self, method): + with pytest.raises(TypeError, match=rf"does not support \.{method}\("): + getattr(SecretStore.api_key, method)("x") + + def test_is_in_raises(self): + with pytest.raises(TypeError, match=r"does not support \.is_in\("): + SecretStore.api_key.is_in(["x", "y"]) # ty: ignore[invalid-argument-type] + + @pytest.mark.parametrize( + "method", ["contains", "icontains", "startswith", "endswith"] + ) + def test_text_pattern_condition_raises(self, method): + """EncryptedTextField inherits TextField's pattern conditions and + blocks them — matching ciphertext by substring is meaningless.""" + with pytest.raises(TypeError, match=rf"does not support \.{method}\("): + getattr(SecretStore.api_key, method)("x") + + def test_every_blocked_method_is_rejected_statically(self): + """Pin the block from the type checker's side, not just the runtime's. + + Each `ty: ignore[invalid-argument-type]` below asserts that the call + is a type error: ty reports an unused suppression as an error of its + own, so if any of these parameters ever widens away from `Never`, + `./scripts/type-check plain-postgres` fails here. That matters because + the `ty: ignore[invalid-method-override]` on EncryptedTextField is + class-wide and would otherwise hide a block that stopped blocking. + + The same calls are asserted to raise, so the runtime and the type + checker are pinned to each other in one place. + """ + with pytest.raises(TypeError): + SecretStore.api_key.equals("x") # ty: ignore[no-matching-overload] + with pytest.raises(TypeError): + SecretStore.api_key.not_equal("x") # ty: ignore[no-matching-overload] + with pytest.raises(TypeError): + SecretStore.api_key.gt("x") # ty: ignore[invalid-argument-type] + with pytest.raises(TypeError): + SecretStore.api_key.gte("x") # ty: ignore[invalid-argument-type] + with pytest.raises(TypeError): + SecretStore.api_key.lt("x") # ty: ignore[invalid-argument-type] + with pytest.raises(TypeError): + SecretStore.api_key.lte("x") # ty: ignore[invalid-argument-type] + with pytest.raises(TypeError): + SecretStore.api_key.is_in(["x"]) # ty: ignore[invalid-argument-type] + with pytest.raises(TypeError): + SecretStore.api_key.contains("x") # ty: ignore[invalid-argument-type] + with pytest.raises(TypeError): + SecretStore.api_key.icontains("x") # ty: ignore[invalid-argument-type] + with pytest.raises(TypeError): + SecretStore.api_key.startswith("x") # ty: ignore[invalid-argument-type] + with pytest.raises(TypeError): + SecretStore.api_key.endswith("x") # ty: ignore[invalid-argument-type] + + def test_json_field_is_rejected_statically(self): + """`config` is annotated `EncryptedField[dict | None]`, so class access + types as the field and the same static block applies. Spelled directly + rather than through getattr, because a marker can't bind to a dynamic + call.""" + with pytest.raises(TypeError): + SecretStore.config.equals({"a": 1}) # ty: ignore[no-matching-overload] + with pytest.raises(TypeError): + SecretStore.config.not_equal({"a": 1}) # ty: ignore[no-matching-overload] + with pytest.raises(TypeError): + SecretStore.config.is_in([{"a": 1}]) # ty: ignore[invalid-argument-type] + + def test_json_field_is_null_survives_the_block(self): + assert_type(SecretStore.config.is_null(), Q) + + def test_is_null_survives_the_block_statically(self): + """The one condition that stays open must keep its real signature - + no ignore marker here, so a `Never` creeping onto is_null breaks the + build.""" + assert_type(SecretStore.api_key.is_null(), Q) + assert_type(SecretStore.api_key.is_null(False), Q) + + @pytest.mark.parametrize( + "method", ["equals", "not_equal", "gt", "gte", "lt", "lte", "is_in"] + ) + def test_json_field_comparison_raises(self, method): + """EncryptedJSONField carries the same block. This case covers the + runtime half; `test_json_field_is_rejected_statically` below covers + the static half, which needs direct call sites rather than getattr.""" + with pytest.raises(TypeError, match=rf"does not support \.{method}\("): + getattr(SecretStore.config, method)("x") + + def test_is_null_returns_correct_lookup(self): + """is_null is the one comparison that makes sense on ciphertext.""" + from plain.postgres.query_utils import Q + + q = SecretStore.api_key.is_null() + assert isinstance(q, Q) + assert q.children == [("api_key__isnull", True)] + + q_false = SecretStore.api_key.is_null(False) + assert q_false.children == [("api_key__isnull", False)] + + +class TestKwargFilterBlocked: + """Block the legacy kwarg/Q path the same way the typed methods are + blocked: `filter(api_key='x')` on an encrypted field would silently + return zero rows because ciphertext is non-deterministic. + `filter(api_key=None)` is preserved so it still rewrites to IS NULL. + """ + + def test_filter_non_none_raises(self, db): + with pytest.raises(TypeError, match=r"api_key.*cannot be matched against"): + SecretStore.query.filter(api_key="sk-test").count() + + def test_exclude_non_none_raises(self, db): + with pytest.raises(TypeError, match=r"api_key.*cannot be matched against"): + SecretStore.query.exclude(api_key="sk-test").count() + + def test_filter_none_still_rewrites_to_isnull(self, db): + """filter(field=None) must continue to work — ORM rewrites to isnull.""" + SecretStore.query.create(name="test", api_key="sk-test", config=None) + assert SecretStore.query.filter(config=None).count() == 1 + + def test_get_or_create_on_an_encrypted_lookup_raises(self, db): + """A deliberate break. This used to "work": ciphertext never matched, + so every call created another row. Raising is the point of the block, + and the message has to say where the value belongs instead.""" + with pytest.raises(TypeError, match=r"move 'api_key' into defaults="): + SecretStore.query.get_or_create(name="test", api_key="sk-test", config=None) + + def test_get_or_create_with_the_encrypted_value_in_defaults_works(self, db): + """The spelling the message points at.""" + obj, created = SecretStore.query.get_or_create( + name="test", defaults={"api_key": "sk-test", "config": None} + ) + assert created + assert obj.api_key == "sk-test" + + again, created_again = SecretStore.query.get_or_create( + name="test", defaults={"api_key": "other", "config": None} + ) + assert not created_again + assert again.id == obj.id + + def test_filter_against_an_expression_raises(self, db): + """An F() right-hand side is a value comparison too — the column is + still ciphertext, so it can never match.""" + with pytest.raises(TypeError, match=r"api_key.*cannot be matched against"): + SecretStore.query.filter(api_key=F("name")).count() + + class TestKeyRotation: def test_decrypt_with_fallback_key(self): """Data encrypted with an old key should decrypt when that key is in fallbacks.""" @@ -217,3 +383,55 @@ def test_literal_default_is_still_rejected(self, value): with pytest.raises(TypeError, match="does not accept a persistent default"): EncryptedJSONField(required=False, allow_null=True, default=value) + + +class TestDeterministicValuesStillMatch: + """The empty string is stored as plaintext `''` (that is what makes + `default=""` expressible as a column DEFAULT), so equality against it is + meaningful and must keep working -- on both the kwarg and the typed path, + which must agree with each other.""" + + def test_filter_on_empty_string_works(self, db): + SecretStore.query.create(name="blank", api_key="k", notes="", config=None) + SecretStore.query.create(name="filled", api_key="k", notes="x", config=None) + + assert SecretStore.query.filter(notes="").count() == 1 + assert SecretStore.query.exclude(notes="").count() == 1 + + def test_get_or_create_on_empty_string_works(self, db): + obj, created = SecretStore.query.get_or_create( + notes="", defaults={"name": "blank", "api_key": "k", "config": None} + ) + assert created + again, created_again = SecretStore.query.get_or_create( + notes="", defaults={"name": "other", "api_key": "k", "config": None} + ) + assert not created_again + assert again.id == obj.id + + def test_typed_equals_matches_the_kwarg_path(self): + """`equals(None)` and `filter(field=None)` can't disagree, and the + error message advertises `=None`, so the typed path has to allow it.""" + assert SecretStore.api_key.equals(None).children == [("api_key", None)] + assert SecretStore.notes.equals("").children == [("notes", "")] + assert SecretStore.notes.not_equal("").children == [("notes", "")] + + def test_where_filters_on_the_empty_string(self, db): + SecretStore.query.create(name="blank", api_key="k", notes="", config=None) + SecretStore.query.create(name="filled", api_key="k", notes="x", config=None) + + rows = list(SecretStore.query.where(SecretStore.notes.equals(""))) + assert [r.name for r in rows] == ["blank"] + + def test_a_real_value_is_still_blocked(self, db): + with pytest.raises(TypeError, match=r"does not support \.equals\("): + SecretStore.notes.equals("something") # ty: ignore[no-matching-overload] + with pytest.raises(TypeError, match=r"cannot be matched against"): + SecretStore.query.filter(notes="something").count() + + def test_json_field_allows_none_but_not_empty_string(self): + """Only text stores "" as plaintext; an empty string on a JSON column + would still be encrypted, so it stays blocked.""" + assert SecretStore.config.equals(None).children == [("config", None)] + with pytest.raises(TypeError, match=r"does not support \.equals\("): + SecretStore.config.equals("") # ty: ignore[no-matching-overload] diff --git a/plain-postgres/tests/public/test_typed_where.py b/plain-postgres/tests/public/test_typed_where.py new file mode 100644 index 0000000000..224dca931e --- /dev/null +++ b/plain-postgres/tests/public/test_typed_where.py @@ -0,0 +1,273 @@ +"""Typed `where()` clause backed by field-method conditions. + +First slice of the typed query API: field descriptors expose `equals`, +`not_equal`, comparison and string lookup methods that return Q objects; +`QuerySet.where()` accepts them positionally. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, assert_type + +import pytest +from app.examples.models.defaults import DefaultsExample +from app.examples.models.string_conditions import StringConditionsExample +from plain.postgres import Field +from plain.postgres.query_utils import Q + + +def test_class_access_yields_typed_descriptors() -> None: + """Class-level field access returns the descriptor, parameterized by T. + + The declared type is what the checker sees — models annotate their fields + `Field[T]`, so that (not the concrete `TextField[T]` the stub returns) is + the type of a class-level access. `Field[T]` therefore has to carry the + whole condition surface, including the string-only conditions, which it + restricts to string-valued fields via the `self` annotation. + + These `assert_type` calls are checked by the type checker, not at runtime + — but the function still has to import cleanly. + """ + assert_type(DefaultsExample.name, Field[str]) + assert_type(DefaultsExample.note, Field[str | None]) + assert_type(DefaultsExample.priority, Field[int]) + + +def test_instance_access_yields_value_type() -> None: + """Instance access returns the value type T (with nullability preserved).""" + row = DefaultsExample(name="x", note=None, priority=1) + assert_type(row.name, str) + assert_type(row.note, str | None) + assert_type(row.priority, int) + + +def test_assignment_typing_accepts_value_type() -> None: + """Assignment to a field instance accepts the declared value type.""" + row = DefaultsExample(name="x", note=None, priority=1) + row.name = "y" # str → str: OK + row.priority = 99 # int → int: OK + row.note = None # None → str | None: OK (nullable) + row.note = "set" # str → str | None: OK + + +if TYPE_CHECKING: + # Type-check only: these assignments must be flagged by ty. The ignore + # markers are load-bearing — if Field.__set__ were typed loosely + # (e.g. value: Any), ty would report them as unused suppressions. + # Their presence here proves the type checker enforces T. We avoid + # running the assignments at runtime because Field.__set__ also calls + # to_python() which raises ValidationError on unconvertible input. + def _typed_check_rejects_wrong_assignment() -> None: + row = DefaultsExample(name="x", note=None, priority=1) + row.name = 123 # ty: ignore[invalid-assignment] + row.priority = "no" # ty: ignore[invalid-assignment] + # Non-nullable field rejects None at type-check time even though the + # runtime would store it (and only fail later at validate/save). + row.name = None # ty: ignore[invalid-assignment] + + def _typed_check_is_in_element_type() -> None: + # is_in takes an iterable of the field's value type. A matching + # iterable type-checks clean; a wrong element type is flagged. The + # ignore marker is load-bearing — if the parameter were typed loosely + # (e.g. Iterable[Any]), ty would report it as an unused suppression. + DefaultsExample.priority.is_in([1, 2, 3]) + DefaultsExample.name.is_in(["a", "b"]) + DefaultsExample.priority.is_in(["no", "ints"]) # ty: ignore[invalid-argument-type] + + def _typed_check_string_conditions_are_string_only() -> None: + # The pattern conditions are declared on Field with a `self` + # annotation that restricts them to string-valued fields, so they + # survive the `Field[T]` annotation models carry without becoming + # available on every field. The ignore marker is load-bearing — if the + # restriction were dropped, ty would report it as unused. + # + # These stay type-check-only because the restriction is type-only: + # `Contains` and friends are registered on `Field` itself, so + # `IntegerField.contains("9")` builds a perfectly valid lookup at + # runtime. The checker is the guard, which is why running these would + # prove nothing. + DefaultsExample.name.startswith("a") + DefaultsExample.note.contains("a") + DefaultsExample.priority.startswith("a") # ty: ignore[invalid-argument-type] + DefaultsExample.priority.contains("a") # ty: ignore[invalid-argument-type] + DefaultsExample.priority.icontains("a") # ty: ignore[invalid-argument-type] + DefaultsExample.priority.endswith("a") # ty: ignore[invalid-argument-type] + + +def test_field_methods_return_q_objects(): + """The methods are usable before any DB hit and produce Q objects.""" + assert isinstance(DefaultsExample.name.equals("foo"), Q) + assert isinstance(DefaultsExample.priority.gte(5), Q) + assert isinstance(DefaultsExample.name.contains("oo"), Q) + assert isinstance(DefaultsExample.note.is_null(), Q) + assert isinstance(DefaultsExample.priority.is_in([1, 2]), Q) + + +def test_where_filters_by_equals(db): + DefaultsExample.query.create(name="alice") + DefaultsExample.query.create(name="bob") + + rows = list(DefaultsExample.query.where(DefaultsExample.name.equals("alice"))) + assert [r.name for r in rows] == ["alice"] + + +def test_where_ands_multiple_conditions(db): + DefaultsExample.query.create(name="alice", priority=1) + DefaultsExample.query.create(name="alice", priority=10) + DefaultsExample.query.create(name="bob", priority=10) + + rows = list( + DefaultsExample.query.where( + DefaultsExample.name.equals("alice"), + DefaultsExample.priority.gte(5), + ) + ) + assert [(r.name, r.priority) for r in rows] == [("alice", 10)] + + +def test_where_combines_with_or(db): + DefaultsExample.query.create(name="alice") + DefaultsExample.query.create(name="bob") + DefaultsExample.query.create(name="carol") + + rows = list( + DefaultsExample.query.where( + DefaultsExample.name.equals("alice") | DefaultsExample.name.equals("carol") + ).order_by("name") + ) + assert [r.name for r in rows] == ["alice", "carol"] + + +def test_not_equal_filters_inverse(db): + DefaultsExample.query.create(name="alice") + DefaultsExample.query.create(name="bob") + + rows = list(DefaultsExample.query.where(DefaultsExample.name.not_equal("alice"))) + assert [r.name for r in rows] == ["bob"] + + +def test_text_field_string_lookups(db): + DefaultsExample.query.create(name="alice") + DefaultsExample.query.create(name="alpha") + DefaultsExample.query.create(name="bob") + + starts = list( + DefaultsExample.query.where(DefaultsExample.name.startswith("al")).order_by( + "name" + ) + ) + assert [r.name for r in starts] == ["alice", "alpha"] + + +def test_where_filters_by_is_in(db): + DefaultsExample.query.create(name="alice") + DefaultsExample.query.create(name="bob") + DefaultsExample.query.create(name="carol") + + rows = list( + DefaultsExample.query.where( + DefaultsExample.name.is_in(["alice", "carol"]) + ).order_by("name") + ) + assert [r.name for r in rows] == ["alice", "carol"] + + +def test_is_in_negation_excludes_members(db): + DefaultsExample.query.create(name="alice") + DefaultsExample.query.create(name="bob") + DefaultsExample.query.create(name="carol") + + rows = list( + DefaultsExample.query.where( + ~DefaultsExample.name.is_in(["alice", "carol"]) + ).order_by("name") + ) + assert [r.name for r in rows] == ["bob"] + + +def test_is_null_with_explicit_default(db): + DefaultsExample.query.create(name="alice", note=None) + DefaultsExample.query.create(name="bob") # default "auto" + + nulls = list(DefaultsExample.query.where(DefaultsExample.note.is_null())) + assert [r.name for r in nulls] == ["alice"] + + non_nulls = list(DefaultsExample.query.where(DefaultsExample.note.is_null(False))) + assert [r.name for r in non_nulls] == ["bob"] + + +# --------------------------------------------------------------------------- +# Pattern conditions live on `Field`, restricted to string-valued fields by +# their `self` annotation. So they reach every string-valued field regardless +# of its base class — `GenericIPAddressField` is a DefaultableField and +# `RandomStringField` is a ColumnField, and neither inherits TextField. +# --------------------------------------------------------------------------- + + +def test_where_filters_ip_address_field_by_pattern(db): + StringConditionsExample.query.create(label="private", ip="10.0.0.1") + StringConditionsExample.query.create(label="local", ip="192.168.1.1") + + rows = list( + StringConditionsExample.query.where( + StringConditionsExample.ip.startswith("10.") + ) + ) + assert [r.label for r in rows] == ["private"] + + +def test_where_filters_random_string_field_by_pattern(db): + """The token is generated by Postgres, so read one back and match on it.""" + StringConditionsExample.query.create(label="only", ip="10.0.0.1") + + token = StringConditionsExample.query.get(label="only").token + assert len(token) == 16 + + rows = list( + StringConditionsExample.query.where( + StringConditionsExample.token.startswith(token[:4]) + ) + ) + assert [r.label for r in rows] == ["only"] + + +def test_pattern_condition_on_a_non_string_field_is_a_static_error() -> None: + """The `self` annotation is the whole guard, and it is a static one — the + load-bearing `ty: ignore` markers in the TYPE_CHECKING block above are what + pin it. + + There is deliberately no runtime counterpart: `Contains`, `IContains`, + `StartsWith` and `EndsWith` are registered on `Field` itself (see + lookups.py), so every field has them and nothing at runtime distinguishes a + string field from an int one. `IntegerField.contains("9")` builds a valid + `priority__contains` lookup that Postgres will happily run. Same type-first + guard model the encrypted fields use. + """ + assert DefaultsExample.priority.get_lookup("contains") is not None + + +# --------------------------------------------------------------------------- +# None operands on ordering conditions. +# +# There is deliberately no static pin here. On a nullable field `T` includes +# None, and None cannot be subtracted from a TypeVar: probed against ty 0.0.80 +# and pyright 1.1.414, `self: Field[X | None], value: X` selects the intended +# overload but solves X as `int | None`, so `.gte(None)` type-checks either +# way. The runtime refusal below is the guard. +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("method", ["gt", "gte", "lt", "lte"]) +def test_ordering_condition_rejects_none(method: str) -> None: + with pytest.raises(TypeError, match=rf"\.{method}\(\) has no meaning for None"): + getattr(DefaultsExample.note, method)(None) + + +def test_ordering_condition_still_accepts_a_value_on_a_nullable_field() -> None: + assert DefaultsExample.note.gte("m").children == [("note__gte", "m")] + + +def test_is_null_and_equals_still_accept_none() -> None: + """`equals(None)` is the ORM's exact-None rewrite and stays legal.""" + assert DefaultsExample.note.equals(None).children == [("note", None)] + assert DefaultsExample.note.is_null().children == [("note__isnull", True)] diff --git a/plain-postgres/tests/public/test_typed_where_fk.py b/plain-postgres/tests/public/test_typed_where_fk.py new file mode 100644 index 0000000000..19c28f614d --- /dev/null +++ b/plain-postgres/tests/public/test_typed_where_fk.py @@ -0,0 +1,361 @@ +"""Typed where() across forward foreign-key relations. + +`ChildCascade.parent` is a ForeignKeyField to DeleteParent. Accessing +`.name` on the class-level descriptor should yield DeleteParent's own `name` +field, renamed to `parent__name`, so its condition methods build that path. +""" + +from __future__ import annotations + +import pytest +from app.examples.models.delete import ( + ChildCascade, + ChildSetNull, + DeleteParent, + Grandchild, + Grandparent, + MidParent, +) +from app.examples.models.relationships import Tag, Widget, WidgetTag +from app.examples.models.shadowing import ShadowSource, ShadowTarget +from plain.postgres.query_utils import Q + + +def test_fk_field_access_builds_prefixed_q(): + q = ChildCascade.parent.name.equals("foo") + assert isinstance(q, Q) + assert q.children == [("parent__name", "foo")] + + +def test_fk_field_access_supports_other_lookups(): + assert ChildCascade.parent.name.startswith("a").children == [ + ("parent__name__startswith", "a") + ] + assert ChildCascade.parent.name.is_null().children == [ + ("parent__name__isnull", True) + ] + assert ChildCascade.parent.name.is_in(["a", "b"]).children == [ + ("parent__name__in", ["a", "b"]) + ] + + +def test_fk_traversal_in_where_clause(db): + """End-to-end: build a query through the FK and verify it runs.""" + parent = DeleteParent.query.create(name="alice") + other = DeleteParent.query.create(name="bob") + ChildCascade.query.create(parent=parent) + ChildCascade.query.create(parent=other) + + matches = list(ChildCascade.query.where(ChildCascade.parent.name.equals("alice"))) + assert len(matches) == 1 + assert matches[0].parent.id == parent.id + + +def test_fk_traversal_combines_with_local_conditions(db): + """Mix a traversal condition with a local condition via &.""" + p1 = DeleteParent.query.create(name="alice") + p2 = DeleteParent.query.create(name="alice") + ChildCascade.query.create(parent=p1) + ChildCascade.query.create(parent=p2) + + matches = list( + ChildCascade.query.where( + ChildCascade.parent.name.equals("alice"), + ChildCascade.id.gte(0), + ) + ) + assert {c.parent.id for c in matches} == {p1.id, p2.id} + + +def test_multiple_fks_on_one_model(db): + """WidgetTag has FKs to both Widget and Tag — each path resolves independently.""" + w = Widget.query.create(name="cog", size="small") + t = Tag.query.create(name="metal") + WidgetTag.query.create(widget=w, tag=t) + + assert WidgetTag.query.where(WidgetTag.widget.name.equals("cog")).count() == 1 + assert WidgetTag.query.where(WidgetTag.tag.name.equals("metal")).count() == 1 + assert WidgetTag.query.where(WidgetTag.widget.name.equals("missing")).count() == 0 + + +def test_fk_traversal_or_combination(db): + p1 = DeleteParent.query.create(name="alice") + p2 = DeleteParent.query.create(name="bob") + p3 = DeleteParent.query.create(name="carol") + ChildCascade.query.create(parent=p1) + ChildCascade.query.create(parent=p2) + ChildCascade.query.create(parent=p3) + + matches = list( + ChildCascade.query.where( + ChildCascade.parent.name.equals("alice") + | ChildCascade.parent.name.equals("carol") + ) + ) + assert {c.parent.name for c in matches} == {"alice", "carol"} + + +def test_unknown_attribute_on_related_raises_attribute_error(): + """Traversal into a non-existent field on the related model fails loudly, + not silently producing a wrong-shaped Q.""" + with pytest.raises(AttributeError): + _ = ChildCascade.parent.nonexistent_field # ty: ignore[unresolved-attribute] + + +# --------------------------------------------------------------------------- +# Multi-hop traversal: Grandchild -> MidParent -> Grandparent +# --------------------------------------------------------------------------- + + +def test_two_hop_traversal_builds_double_prefixed_q(): + q = Grandchild.mid_parent.grandparent.name.equals("alice") + assert q.children == [("mid_parent__grandparent__name", "alice")] + + +def test_two_hop_traversal_runs(db): + g1 = Grandparent.query.create(name="alice") + g2 = Grandparent.query.create(name="bob") + m1 = MidParent.query.create(grandparent=g1) + m2 = MidParent.query.create(grandparent=g2) + Grandchild.query.create(mid_parent=m1) + Grandchild.query.create(mid_parent=m2) + + matches = list( + Grandchild.query.where(Grandchild.mid_parent.grandparent.name.equals("alice")) + ) + assert {gc.mid_parent.grandparent.name for gc in matches} == {"alice"} + + +def test_two_hop_chain_combines_with_or(db): + g1 = Grandparent.query.create(name="alice") + g2 = Grandparent.query.create(name="bob") + g3 = Grandparent.query.create(name="carol") + m1 = MidParent.query.create(grandparent=g1) + m2 = MidParent.query.create(grandparent=g2) + m3 = MidParent.query.create(grandparent=g3) + Grandchild.query.create(mid_parent=m1) + Grandchild.query.create(mid_parent=m2) + Grandchild.query.create(mid_parent=m3) + + matches = list( + Grandchild.query.where( + Grandchild.mid_parent.grandparent.name.equals("alice") + | Grandchild.mid_parent.grandparent.name.startswith("c") + ) + ) + assert {gc.mid_parent.grandparent.name for gc in matches} == {"alice", "carol"} + + +# --------------------------------------------------------------------------- +# Encrypted field traversal — comparison must be rejected, mirroring the +# direct-access behavior added in the previous commit. +# --------------------------------------------------------------------------- + + +# --------------------------------------------------------------------------- +# Descriptor attribute shadowing: these four names were once public attributes +# on ForwardForeignKeyDescriptor, so a related field named after one of them +# resolved to the descriptor's attribute instead of traversing. They are +# `_`-prefixed now, and `Meta` skips `_`-prefixed attributes when it collects +# fields, so no field name can ever collide with a descriptor attribute again. +# These cases pin that: re-publishing any of them would fail here. +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "name", + ["field", "is_cached", "get_queryset", "get_prefetch_queryset"], +) +def test_shadowed_field_name_traverses_to_field(name): + """ShadowTarget defines fields named after descriptor attributes. + Traversal through ShadowSource.ref resolves each to the field, building + a `ref__` path rather than returning the descriptor's attribute.""" + ref = getattr(ShadowSource.ref, name) + q = ref.equals("x") + assert q.children == [(f"ref__{name}", "x")] + + +def test_shadowed_field_traversal_runs(db): + """End-to-end: a where() through the shadowed `field` name filters rows.""" + matched = ShadowTarget.query.create( + field="hit", + is_cached="a", + get_queryset="b", + get_prefetch_queryset="c", + ) + ShadowTarget.query.create( + field="miss", + is_cached="a", + get_queryset="b", + get_prefetch_queryset="c", + ) + ShadowSource.query.create(ref=matched) + + rows = list(ShadowSource.query.where(ShadowSource.ref.field.equals("hit"))) + assert [r.ref.id for r in rows] == [matched.id] + + +# --------------------------------------------------------------------------- +# Conditions on the relation itself. +# +# A relation is not a field, so it has no condition methods -- and it can't be +# given any. To the type checker `ChildCascade.parent` is `type[DeleteParent]` +# (Field.__get__'s model-valued overloads), which is exactly what lets chained +# traversal type-check; a runtime `.equals()` the checker rejects would be +# worse than none. The spelling is traversal to the key the relation targets, +# which compiles to the same lookup `filter(parent=obj)` produces. +# --------------------------------------------------------------------------- + + +def test_relation_key_conditions_build_q(): + assert ChildCascade.parent.id.equals(7).children == [("parent__id", 7)] + assert ChildCascade.parent.id.is_in([1, 2]).children == [("parent__id__in", [1, 2])] + assert ChildSetNull.parent.id.is_null().children == [("parent__id__isnull", True)] + assert ChildCascade.parent.id.not_equal(7).children == [("parent__id", 7)] + + +def test_where_filters_by_relation_key(db): + kept = DeleteParent.query.create(name="kept") + other = DeleteParent.query.create(name="other") + mine = ChildCascade.query.create(parent=kept) + ChildCascade.query.create(parent=other) + + rows = list(ChildCascade.query.where(ChildCascade.parent.id.equals(kept.id))) + assert [r.id for r in rows] == [mine.id] + + +def test_relation_key_matches_filter_on_the_relation(db): + """The documented equivalence: traversing to the key is the typed spelling + of `filter(parent=obj)`, not merely something similar.""" + kept = DeleteParent.query.create(name="kept") + other = DeleteParent.query.create(name="other") + ChildCascade.query.create(parent=kept) + ChildCascade.query.create(parent=other) + + typed = [ + r.id for r in ChildCascade.query.where(ChildCascade.parent.id.equals(kept.id)) + ] + untyped = [r.id for r in ChildCascade.query.filter(parent=kept)] + assert typed == untyped + assert len(typed) == 1 + + +def test_where_filters_by_relation_key_is_in(db): + a = DeleteParent.query.create(name="a") + b = DeleteParent.query.create(name="b") + c = DeleteParent.query.create(name="c") + for parent in (a, b, c): + ChildCascade.query.create(parent=parent) + + rows = list(ChildCascade.query.where(ChildCascade.parent.id.is_in([a.id, c.id]))) + assert sorted(r.parent.id for r in rows) == sorted([a.id, c.id]) + + +def test_where_filters_by_null_relation_key(db): + kept = DeleteParent.query.create(name="kept") + doomed = DeleteParent.query.create(name="doomed") + attached = ChildSetNull.query.create(parent=kept) + orphan = ChildSetNull.query.create(parent=doomed) + + # The field is nullable but still `required`, so the null arrives the way + # it does in practice: SET_NULL clearing the key when the parent goes. + DeleteParent.query.filter(id=doomed.id).delete() + + nulls = list(ChildSetNull.query.where(ChildSetNull.parent.id.is_null())) + assert [r.id for r in nulls] == [orphan.id] + + non_nulls = list(ChildSetNull.query.where(ChildSetNull.parent.id.is_null(False))) + assert [r.id for r in non_nulls] == [attached.id] + + +@pytest.mark.parametrize("method", ["equals", "is_in", "is_null", "gte", "contains"]) +def test_condition_on_the_relation_itself_raises_helpful_error(method): + """An AttributeError, so `hasattr`/`getattr(..., default)` keep working -- + but one that names the spelling that does work, or the constraint just + looks like a missing feature. (The sweep across every condition name is in + tests/internal/test_typed_where_internals.py.)""" + with pytest.raises(AttributeError) as excinfo: + getattr(ChildCascade.parent, method) + + message = str(excinfo.value) + assert "is a relation, not a field" in message + assert f"parent.id.{method}(...)" in message + + +def test_condition_on_the_relation_keeps_the_attribute_protocol(): + assert not hasattr(ChildCascade.parent, "equals") + assert getattr(ChildCascade.parent, "equals", "fallback") == "fallback" + + +def test_unknown_relation_attribute_still_raises_attribute_error(): + """A genuine typo gets the plain "not a traversable field" message rather + than the condition-method advice.""" + with pytest.raises(AttributeError, match="parent.nope is not a traversable"): + getattr(ChildCascade.parent, "nope") + + +def test_related_field_named_like_a_condition_still_traverses(): + """The field lookup runs first, so a related model that really does have a + column named after a condition method still resolves to the column.""" + assert ShadowSource.ref.field.equals("hit").children == [("ref__field", "hit")] + + +# --------------------------------------------------------------------------- +# Many-to-many relations traverse like any other. Reached *through* a relation, +# `widget__tags__name` is as valid a lookup path as `widget__author__name`, so +# an M2M is a hop, not a leaf -- handing the M2M field back renamed would make +# `.name` resolve to the string "widget__tags". +# +# (Direct class access, `Widget.tags`, is a ForwardManyToManyDescriptor and has +# no traversal wiring; this branch covers forward-FK traversal only.) +# +# This hop is runtime-only for now: at the type level `tags` is declared +# `ManyToManyManager[Tag]`, which exposes the manager API, not the target +# model's fields. Typing it would mean claiming class access to an M2M yields +# `type[Tag]` -- true after a traversal hop, false for `Widget.tags` itself -- +# so the ignores below are the honest marker rather than a papered-over bug. +# --------------------------------------------------------------------------- + + +def test_m2m_traversal_through_a_foreign_key(): + q = WidgetTag.widget.tags.name.equals( # ty: ignore[unresolved-attribute] + "metal" + ) + assert q.children == [("widget__tags__name", "metal")] + + +def test_condition_on_a_traversed_m2m_gets_the_same_advice(): + with pytest.raises(AttributeError) as excinfo: + getattr(WidgetTag.widget.tags, "equals") + + message = str(excinfo.value) + assert "is a relation, not a field" in message + assert "WidgetTag.widget.tags" not in message # the prefix, not the model + assert "widget.tags.id.equals(...)" in message + + +def test_where_filters_through_a_foreign_key_then_an_m2m(db): + metal = Tag.query.create(name="metal") + plastic = Tag.query.create(name="plastic") + cog = Widget.query.create(name="cog", size="small") + knob = Widget.query.create(name="knob", size="small") + WidgetTag.query.create(widget=cog, tag=metal) + WidgetTag.query.create(widget=knob, tag=plastic) + + condition = WidgetTag.widget.tags.name.equals( # ty: ignore[unresolved-attribute] + "metal" + ) + rows = list(WidgetTag.query.where(condition)) + assert [r.widget.id for r in rows] == [cog.id] + + +def test_reverse_relation_says_it_is_not_traversable(): + """`filter(parent__childcascade_set__...)` works, so "not a traversable + field or relation" would be a lie -- the typed API is what can't express + it, and the message has to say which.""" + with pytest.raises(AttributeError) as excinfo: + getattr(ChildCascade.parent, "childcascade_set") + + message = str(excinfo.value) + assert "reverse relation" in message + assert "filter(parent__childcascade_set__...=...)" in message