diff --git a/AGENTS.md b/AGENTS.md index 007249fe..d87ca665 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -246,6 +246,15 @@ Phantom diffs are the canonical symptom that this invariant has been broken. pin the names (`test_index_name_matches_rust_runtime_*`). - `docs/solutions/patterns/cross-emitter-ddl-parity.md` documents the rule and the recipes below. +- On the query side, the rendered-statement parity pin in + `tests/test_rendered_statement.py` is the counterpart of the cross-emitter + pins: for every `Query` terminal (`all`, `first`, `count`, `exists`, + `update` in both forms, `delete`, through `Relation` and `ProjectedQuery` + too) on both dialects, with a secret column, `await query..sql()` + equals the `ferro.sql` record the executed call writes in `sql`, `params` + (the `REDACTED` sentinel included) and `dialect`. Both doors build the + statement through one `build_` function in `src/operations.rs`, + which the executing entry sends and its `render_` twin returns. ### Adding a new emitter diff --git a/docs/pages/api/queries.md b/docs/pages/api/queries.md index 1928584c..265673cd 100644 --- a/docs/pages/api/queries.md +++ b/docs/pages/api/queries.md @@ -37,6 +37,12 @@ An aggregate-only projection collapses to one record (read with `first()`). Mixi ::: ferro.query.builder.ProjectedQuery +::: ferro.RenderedSql + +::: ferro.REDACTED + +::: ferro.query.rendered.Terminal + ::: ferro.query.Row ::: ferro.query.Rows diff --git a/src/ferro/__init__.py b/src/ferro/__init__.py index 1e540208..21c1b644 100644 --- a/src/ferro/__init__.py +++ b/src/ferro/__init__.py @@ -50,7 +50,7 @@ _carrying_report, _run_pass, ) -from .query import Relation, Row, Rows, now +from .query import REDACTED, Relation, RenderedSql, Row, Rows, now from .raw import Transaction, execute, fetch_all, fetch_one from .rowsecurity import RowPolicy, RowSecurity from .session import Session, current_session, engines @@ -587,6 +587,8 @@ async def migrate( "Relation", "Row", "Rows", + "RenderedSql", + "REDACTED", "version", "create_tables", "migrate", diff --git a/src/ferro/_core.pyi b/src/ferro/_core.pyi index a5bab693..e1b2251b 100644 --- a/src/ferro/_core.pyi +++ b/src/ferro/_core.pyi @@ -707,6 +707,42 @@ async def update_filtered( *, secret_values: list[object] = ..., ) -> int: ... + +RenderedParts = tuple[str, tuple[object, ...], tuple[bool, ...], str] +"""A rendering entry's answer: the SQL, the parameters (`None` where +masked), the mask (`True` where the parameter is a secret) and the dialect.""" + +async def render_fetch( + cls: object, + query_ir_json: str, + route: RouteHandle, + record_cls: type | None = None, + hop_classes: dict[str, type] | None = None, + *, + secret_values: list[object] = ..., +) -> RenderedParts: ... +async def render_count( + name: str, + query_ir_json: str, + route: RouteHandle, + *, + secret_values: list[object] = ..., + operation: Literal["count", "exists"] = "count", +) -> RenderedParts: ... +async def render_update( + name: str, + query_ir_json: str, + route: RouteHandle, + *, + secret_values: list[object] = ..., +) -> RenderedParts: ... +async def render_delete( + name: str, + query_ir_json: str, + route: RouteHandle, + *, + secret_values: list[object] = ..., +) -> RenderedParts: ... async def add_m2m_links( join_table: str, source_col: str, diff --git a/src/ferro/query/__init__.py b/src/ferro/query/__init__.py index 355d97c7..56123300 100644 --- a/src/ferro/query/__init__.py +++ b/src/ferro/query/__init__.py @@ -12,9 +12,11 @@ ValueExpr, now, ) +from .rendered import REDACTED, RenderedSql from .rows import Row, Rows __all__ = [ + "REDACTED", "AggregateExpr", "FieldProxy", "Predicate", @@ -24,6 +26,7 @@ "QueryProxy", "Relation", "RelationProxy", + "RenderedSql", "Row", "RowSelector", "Rows", diff --git a/src/ferro/query/builder.py b/src/ferro/query/builder.py index 16dc2c09..6ac11d6f 100644 --- a/src/ferro/query/builder.py +++ b/src/ferro/query/builder.py @@ -7,15 +7,18 @@ """ import copy +from collections.abc import Awaitable from typing import ( TYPE_CHECKING, Any, Callable, + ClassVar, Generic, Literal, NamedTuple, Never, NoReturn, + Protocol, Self, Type, TypeVar, @@ -30,6 +33,10 @@ delete_filtered, fetch_filtered, remove_m2m_links, + render_count, + render_delete, + render_fetch, + render_update, update_filtered, ) from .nodes import ( @@ -44,6 +51,7 @@ ValueExpr, validate_query_column, ) +from .rendered import Terminal, send, terminal from .rows import Row, Rows from .wire import ( M2mContext, @@ -554,6 +562,48 @@ def _collect_dict_projection(result: dict[Any, Any]) -> tuple[_ProjectedField, . return tuple(fields) +if TYPE_CHECKING: + from .rendered import RenderedSql + + class _UpdateTerminal(Protocol[E]): + """`query.update` as a type checker reads it: a recipe or keyword + literals, never both, on the call and on `.sql()` alike.""" + + @overload + def __call__( + self, recipe: Callable[[QueryProxy[E]], dict[str, Any]], / + ) -> Awaitable[int]: ... + @overload + def __call__(self, /, **fields: Any) -> Awaitable[int]: ... + @overload + def sql( + self, recipe: Callable[[QueryProxy[E]], dict[str, Any]], / + ) -> Awaitable[RenderedSql]: ... + @overload + def sql(self, /, **fields: Any) -> Awaitable[RenderedSql]: ... + + class _UpdateMethod(Protocol): + """`Query.update` as a type checker reads it: reading it from a + `Query[E]` binds an `_UpdateTerminal[E]`.""" + + @overload + def __get__(self, query: None, owner: Any) -> "_UpdateMethod": ... + @overload + def __get__(self, query: "Query[E]", owner: Any) -> _UpdateTerminal[E]: ... + + +def _update_terminal(method: Callable[..., Awaitable[int]]) -> "_UpdateMethod": + """Make `Query.update` a [`terminal`][ferro.query.rendered.terminal] that + keeps its two forms apart statically. + + A terminal carries its method's one signature, and `update` has two + (a recipe, or keyword literals), so a type checker reads this terminal + through `_UpdateMethod`, whose call and `.sql()` are overloaded per form. + At runtime it is `terminal(method)`. + """ + return cast("_UpdateMethod", terminal(method)) + + class Query(Generic[T]): """Build and execute fluent ORM queries. @@ -1236,7 +1286,8 @@ def before(self, position: tuple[Any, ...] | T) -> Self: new._before = bound return new - async def all(self) -> list[T]: + @terminal + async def all(self: "Query[T]") -> list[T]: """Return all model instances that match the current query Returns: @@ -1249,7 +1300,7 @@ async def all(self) -> list[T]: """ compiled = compile_query(self, "fetch") route = await self._transaction_or_using() - rows = await fetch_filtered( + rows = await send(fetch_filtered, render_fetch)( self.model_cls, compiled.wire_json, route, @@ -1260,7 +1311,8 @@ async def all(self) -> list[T]: rows.reverse() return rows - async def count(self) -> int: + @terminal + async def count(self: "Query[T]") -> int: """Return the number of records that match the current query Returns: @@ -1277,7 +1329,7 @@ async def _count(self, operation: Literal["count", "exists"]) -> int: """Count the matching rows, naming the statement-log records `operation`.""" compiled = compile_query(self, "count") route = await self._transaction_or_using() - return await count_filtered( + return await send(count_filtered, render_count)( model_identity(self.model_cls), compiled.wire_json, route, @@ -1285,15 +1337,13 @@ async def _count(self, operation: Literal["count", "exists"]) -> int: operation=operation, ) - @overload + @_update_terminal async def update( - self, recipe: Callable[[QueryProxy[T]], dict[str, Any]], / - ) -> int: ... - - @overload - async def update(self, /, **fields: Any) -> int: ... - - async def update(self, *args: Any, **fields: Any) -> int: + self: "Query[T]", + recipe: Callable[[QueryProxy[T]], dict[str, Any]] | None = None, + /, + **fields: Any, + ) -> int: """Update all records matching the current query. Two doors, one SET wire: keyword literals (``update(name="x")``) or a @@ -1306,10 +1356,10 @@ async def update(self, *args: Any, **fields: Any) -> int: cannot be mixed. Args: - *args: Optional positional recipe callable receiving a + recipe: Optional positional recipe callable receiving a :class:`QueryProxy` and returning a non-empty ``dict`` of - root-column assignments. Nameless so a keyword ``recipe=`` - still SETs a column of that name. + root-column assignments. Positional-only, so a keyword + ``recipe=`` still SETs a column of that name. **fields: Field names and literal values to update. Returns: @@ -1335,9 +1385,6 @@ async def update(self, *args: Any, **fields: Any) -> int: ... } ... ) """ - if len(args) > 1: - raise TypeError("update() takes at most one positional recipe") - recipe = args[0] if args else None if recipe is not None and fields: raise TypeError( "update() accepts either a recipe callable or keyword " @@ -1350,14 +1397,15 @@ async def update(self, *args: Any, **fields: Any) -> int: else: compiled = compile_query(self, "update", assignments=fields) route = await self._transaction_or_using() - return await update_filtered( + return await send(update_filtered, render_update)( model_identity(self.model_cls), compiled.wire_json, route, secret_values=list(compiled.secret_values), ) - async def first(self) -> T | None: + @terminal + async def first(self: "Query[T]") -> T | None: """Return the first matching record, or None Returns: @@ -1371,7 +1419,8 @@ async def first(self) -> T | None: results = await self.limit(1).all() return results[0] if results else None - async def delete(self) -> int: + @terminal + async def delete(self: "Query[T]") -> int: """Delete all records matching the current query Returns: @@ -1392,14 +1441,15 @@ async def delete(self) -> int: """ compiled = compile_query(self, "delete") route = await self._transaction_or_using() - return await delete_filtered( + return await send(delete_filtered, render_delete)( model_identity(self.model_cls), compiled.wire_json, route, secret_values=list(compiled.secret_values), ) - async def exists(self) -> bool: + @terminal + async def exists(self: "Query[T]") -> bool: """Return whether at least one record matches the current query Returns: @@ -1724,7 +1774,8 @@ def _is_group_key_source(self, column: str, path: tuple[str, ...]) -> bool: for field in self._projection ) - def count(self): # type: ignore[override] + @terminal + def count(self: "ProjectedQuery[T]") -> Awaitable[int]: """Count matching rows — or raise on an aggregate projection (#295). On a plain projection ``count()`` stays projection-blind (#279's @@ -1745,7 +1796,8 @@ def count(self): # type: ignore[override] ) return super().count() - def exists(self): # type: ignore[override] + @terminal + def exists(self: "ProjectedQuery[T]") -> Awaitable[bool]: """Existence check — or raise on an aggregate projection (#295). Raises: @@ -1808,7 +1860,8 @@ def _read_projected_order_value(self, row: Row, entry: OrderByEntry) -> Any: ) return getattr(row, name) - async def all(self) -> Rows[Row]: # type: ignore[override] # ty: ignore[invalid-method-override] + @terminal + async def all(self: "ProjectedQuery[T]") -> Rows[Row]: # type: ignore[override] """Return the projected records for every matching row. Returns: @@ -1822,7 +1875,7 @@ async def all(self) -> Rows[Row]: # type: ignore[override] # ty: ignore[invali """ compiled = compile_query(self, "fetch") route = await self._transaction_or_using() - records = await fetch_filtered( + records = await send(fetch_filtered, render_fetch)( self.model_cls, compiled.wire_json, route, @@ -1834,7 +1887,8 @@ async def all(self) -> Rows[Row]: # type: ignore[override] # ty: ignore[invali records.reverse() return _ROWS_OF_ROW._wrap(records) - async def first(self) -> Row | None: # type: ignore[override] # ty: ignore[invalid-method-override] + @terminal + async def first(self: "ProjectedQuery[T]") -> Row | None: # type: ignore[override] """Return the first matching projected record, or None. Examples: @@ -1931,6 +1985,41 @@ def __repr__(self): ) +if TYPE_CHECKING: + + class _RelationAll: + """`Relation.all` as a type checker reads it: its rows are instances + of `E` whether the relation is written `Relation[list[E]]` or + `Relation[E]`.""" + + @overload + def __get__(self, query: None, owner: Any) -> "_RelationAll": ... + @overload + def __get__( + self, query: "Relation[list[E]]", owner: Any + ) -> Terminal[[], list[E]]: ... + @overload + def __get__( + self, query: "Relation[E]", owner: Any + ) -> Terminal[[], list[E]]: ... + def __get__(self, query: Any, owner: Any) -> Any: ... + + class _RelationFirst: + """`Relation.first` as a type checker reads it (see `_RelationAll`).""" + + @overload + def __get__(self, query: None, owner: Any) -> "_RelationFirst": ... + @overload + def __get__( + self, query: "Relation[list[E]]", owner: Any + ) -> Terminal[[], E | None]: ... + @overload + def __get__( + self, query: "Relation[E]", owner: Any + ) -> Terminal[[], E | None]: ... + def __get__(self, query: Any, owner: Any) -> Any: ... + + class Relation(Query[T]): """Represent lazy collection relationship queries with typing support @@ -1961,24 +2050,8 @@ class Relation(Query[T]): # Relation[T] as a query whose *rows* are model instances, regardless of whether # T is written as Model or list[Model] in the field annotation. if TYPE_CHECKING: - - @overload - async def all(self: "Relation[list[E]]") -> list[E]: ... - - @overload - async def all(self: "Relation[E]") -> list[E]: ... - - @overload - async def first(self: "Relation[list[E]]") -> E | None: ... - - @overload - async def first(self: "Relation[E]") -> E | None: ... - - async def all(self): # type: ignore[override] - return await super().all() - - async def first(self): # type: ignore[override] - return await super().first() + all: ClassVar["_RelationAll"] # type: ignore[assignment] + first: ClassVar["_RelationFirst"] # type: ignore[assignment] @classmethod def __get_pydantic_core_schema__(cls, _source_type, _handler): diff --git a/src/ferro/query/rendered.py b/src/ferro/query/rendered.py new file mode 100644 index 00000000..1c087d7f --- /dev/null +++ b/src/ferro/query/rendered.py @@ -0,0 +1,232 @@ +"""Rendered statements: what a query terminal would send, without sending it. + +A terminal (`all`, `first`, `count`, `exists`, `update`, `delete`) is an +`async def` method wrapped by [`terminal`][ferro.query.rendered.terminal]. +Calling it runs the method; its `.sql` attribute runs the same method in +render mode, where the one engine call that would send the statement is +swapped for its rendering twin. Both twins build the statement through the +same engine function, so `.sql()` returns exactly what the call sends. +""" + +from __future__ import annotations + +import functools +from collections.abc import Awaitable, Callable +from contextvars import ContextVar +from dataclasses import dataclass +from typing import ( + TYPE_CHECKING, + Any, + Concatenate, + Generic, + Literal, + ParamSpec, + TypeVar, + overload, +) + +from .._sql_log import REDACTED + +if TYPE_CHECKING: + from typing import Self + +__all__ = ["REDACTED", "RenderedSql", "Terminal", "TerminalMethod", "send", "terminal"] + +S = TypeVar("S") +P = ParamSpec("P") +R = TypeVar("R") +R_co = TypeVar("R_co", covariant=True) + +_RenderedParts = tuple[str, tuple[object, ...], tuple[bool, ...], str] +"""What an engine rendering entry returns: the SQL, its parameters (`None` +where masked), the mask (`True` where the parameter is a secret) and the +dialect.""" + + +@dataclass(frozen=True, slots=True) +class RenderedSql: + """One statement a query terminal would send, as the statement log records it. + + `str()` of it is the SQL alone, so `print(rendered)` shows the statement. + + Examples: + ```python {test=run prelude=reference} + rendered = await Author.where(lambda author: author.id == 1).all.sql() + print(rendered.dialect, rendered.params) + ``` + """ + + sql: str + """The SQL text, with the dialect's placeholders (`$1` or `?`).""" + params: tuple[object, ...] + """The bound values in placeholder order; a secret value is + [`REDACTED`][ferro.REDACTED].""" + dialect: Literal["postgres", "sqlite"] + """The dialect of the database the call would run on.""" + + def __str__(self) -> str: + """Return the SQL alone.""" + return self.sql + + +_DIALECTS: dict[str, Literal["postgres", "sqlite"]] = { + "postgres": "postgres", + "sqlite": "sqlite", +} +"""The dialect names the engine renders for, as `RenderedSql.dialect` types them.""" + +_rendering: ContextVar[bool] = ContextVar("ferro_rendering", default=False) +"""Whether the running terminal is rendering (`.sql()`) rather than sending.""" + + +class _Rendered(Exception): + """Carries a terminal's rendered statement out of its method body. + + Raised where the method would have sent its statement, so nothing after + the send (decoding, the identity map, the result's post-processing) runs + when rendering. + """ + + def __init__(self, rendered: RenderedSql) -> None: + super().__init__(rendered.sql) + self.rendered = rendered + + +def send( + execute: Callable[P, Awaitable[R]], + render: Callable[P, Awaitable[_RenderedParts]], +) -> Callable[P, Awaitable[R]]: + """Pair an engine entry that sends a statement with its rendering twin. + + A terminal's method sends its statement through this pair. Called + normally, it runs `execute`. Called from `.sql()`, it runs `render` with + the same arguments and hands the rendered statement back to `.sql()`. + + Args: + execute: The engine entry that builds and sends the statement. + render: The engine entry that builds the same statement and returns + it without sending it. + + Returns: + A callable with `execute`'s signature. + """ + + @functools.wraps(execute) + async def call(*args: P.args, **kwargs: P.kwargs) -> R: + if not _rendering.get(): + return await execute(*args, **kwargs) + sql, params, mask, dialect = await render(*args, **kwargs) + raise _Rendered( + RenderedSql( + sql=sql, + params=tuple( + REDACTED if masked else value + for value, masked in zip(params, mask, strict=True) + ), + dialect=_DIALECTS[dialect], + ) + ) + + return call + + +class Terminal(Generic[P, R_co]): + """A query terminal bound to its query: awaiting a call runs it, `.sql()` renders it.""" + + __slots__ = ("_method", "_query") + + def __init__(self, method: Callable[..., Awaitable[R_co]], query: object) -> None: + self._method = method + self._query = query + + def __call__(self, *args: P.args, **kwargs: P.kwargs) -> Awaitable[R_co]: + """Run the terminal: send its statement and return its result.""" + return self._method(self._query, *args, **kwargs) + + async def sql(self, *args: P.args, **kwargs: P.kwargs) -> RenderedSql: + """Render the statement this terminal would send, without sending it. + + Takes exactly the arguments the terminal takes and routes like the + call: the enclosing session, `using=` and `session=` pick the + database. On Postgres it may read the table's column types once, the + way the call would; it never sends the statement itself, opens no + transaction and changes no row. + + Args: + *args: The terminal's positional arguments. + **kwargs: The terminal's keyword arguments. + + Returns: + The SQL, parameters and dialect the call would send, as the + statement log would record them, with each secret value shown + as [`REDACTED`][ferro.REDACTED]. + + Raises: + RuntimeError: No database route: the call is outside a session + and no connection was named. + TypeError: The arguments are ones the terminal rejects. + + Examples: + ```python {test=run prelude=reference} + rendered = await Author.where(lambda author: author.id == 1).all.sql() + print(rendered) + ``` + """ + token = _rendering.set(True) + try: + await self._method(self._query, *args, **kwargs) + except _Rendered as rendered: + return rendered.rendered + finally: + _rendering.reset(token) + raise RuntimeError( + f"{getattr(self._method, '__qualname__', self._method)!r} returned " + "without sending a statement, so there is nothing to render" + ) + + def __repr__(self) -> str: + return f"" + + +class TerminalMethod(Generic[S, P, R_co]): + """A query terminal on its class: reading it from a query binds a [`Terminal`][ferro.query.rendered.Terminal].""" + + def __init__(self, method: Callable[Concatenate[S, P], Awaitable[R_co]]) -> None: + self._method = method + functools.update_wrapper(self, method) + + @overload + def __get__(self, query: None, owner: type[Any]) -> Self: ... + + @overload + def __get__(self, query: S, owner: type[Any]) -> Terminal[P, R_co]: ... + + def __get__(self, query: S | None, owner: type[Any]) -> Self | Terminal[P, R_co]: + if query is None: + return self + return Terminal(self._method, query) + + def __call__(self, query: S, *args: P.args, **kwargs: P.kwargs) -> Awaitable[R_co]: + """Run the terminal on `query`, read from the class.""" + return self._method(query, *args, **kwargs) + + +def terminal( + method: Callable[Concatenate[S, P], Awaitable[R]], +) -> TerminalMethod[S, P, R]: + """Make a query method a terminal: callable to run, `.sql()` to render. + + The method sends its statement through [`send`][ferro.query.rendered.send]; + `.sql()` runs the same method with that send swapped for its rendering + twin, so the call and the rendering share every argument check, the + routing and the statement build. + + Args: + method: The `async def` query method, or a method returning another + terminal's awaitable. + + Returns: + The terminal, read from a query as a bound + [`Terminal`][ferro.query.rendered.Terminal]. + """ + return TerminalMethod(method) diff --git a/src/lib.rs b/src/lib.rs index f9905af9..7c34b0b6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -135,6 +135,10 @@ fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_function(wrap_pyfunction!(operations::fetch_all, m)?)?; m.add_function(wrap_pyfunction!(operations::fetch_filtered, m)?)?; m.add_function(wrap_pyfunction!(operations::count_filtered, m)?)?; + m.add_function(wrap_pyfunction!(operations::render_fetch, m)?)?; + m.add_function(wrap_pyfunction!(operations::render_count, m)?)?; + m.add_function(wrap_pyfunction!(operations::render_update, m)?)?; + m.add_function(wrap_pyfunction!(operations::render_delete, m)?)?; m.add_function(wrap_pyfunction!(operations::fetch_one, m)?)?; m.add_function(wrap_pyfunction!(operations::register_instance, m)?)?; m.add_function(wrap_pyfunction!(operations::evict_instance, m)?)?; diff --git a/src/operations.rs b/src/operations.rs index 1584a94d..c94b459d 100644 --- a/src/operations.rs +++ b/src/operations.rs @@ -1382,6 +1382,24 @@ fn engine_bind_values_from_sea(values: &[SeaValue]) -> Vec { .collect() } +/// One statement as an operation builds it: the SQL and its binds in +/// placeholder order. What an executing entry sends and its rendering twin +/// returns, from the same `build_*` function. +#[derive(Debug, Clone, PartialEq)] +struct BuiltStatement { + sql: String, + binds: Vec, +} + +impl BuiltStatement { + fn from_sea(sql: String, values: &sea_query::Values) -> Self { + Self { + sql, + binds: engine_bind_values_from_sea(&values.0), + } + } +} + /// The statement-log context an operation entry opens (ADR-0057), or `None` /// when `ferro.sql` is not enabled for `DEBUG`. Asked once, while the entry /// holds the GIL; `secret_values` is called only when the log is on, so an @@ -3275,45 +3293,30 @@ mod bulk_chunking_tests { } } -/// Fetches records for a given model class based on a QueryIR-defined query. -/// -/// The query's materialization plan (ADR-0007) selects the result shape: -/// `root_instances` hydrates complete model instances (identity-map aware); -/// a `record` plan renders the projected SELECT list and hydrates -/// `record_cls` records (#279) — no identity map, no persistence identity; -/// an `instances` plan widens the SELECT with every included hop's complete -/// aliased row and hydrates a populated instance graph (#286, ADR-0008). -/// -/// Args: -/// cls (PyAny): The Python model class. -/// query_ir_json (str): The serialized QueryIR envelope JSON. -/// record_cls (PyAny | None): Record constructor for a `record` plan -/// (`Row` today; the seam a future `into=` plugs into). Required for -/// a record plan, forbidden otherwise. -/// hop_classes (dict[str, type] | None): Included hop model classes keyed -/// by physical table name, for an `instances` plan. Required for an -/// instances plan, forbidden otherwise. -/// -/// Returns: -/// list[PyAny]: Hydrated model instances (populated when included), or -/// projected records. -#[pyfunction] -#[pyo3(signature = ( - cls, query_ir_json, route, record_cls=None, hop_classes=None, *, secret_values=Vec::new() -))] -pub fn fetch_filtered<'py>( - py: Python<'py>, - cls: Bound<'py, PyAny>, - query_ir_json: String, - route: Py, - record_cls: Option>, - hop_classes: Option>, - secret_values: Vec>, -) -> PyResult> { - let name = crate::state::model_identity(&cls)?; - let cls_py = cls.unbind(); +/// What a fetch carries besides its statement: the plan's include paths +/// and projected fields, and the record constructor and hop classes the +/// call passed, checked against the plan. Decided once, here, for the +/// executing entry and its rendering twin alike, so both refuse the same +/// calls with the same errors. +struct FetchRequest { + include_paths: Option>>, + projected: Option>, + record_cls_py: Option>, + hop_classes_py: Option>, +} - let mut plan = query_plan_from_ir_json(&query_ir_json)?; +/// Parse and check a fetch's plan, and pair the call's `record_cls` and +/// `hop_classes` with it. +/// +/// # Errors +/// `PyValueError` for a malformed or unsupported plan, or a constructor or +/// hop-class map the plan does not call for (or lacks). +fn fetch_request( + query_ir_json: &str, + record_cls: Option>, + hop_classes: Option>, +) -> PyResult<(QueryPlan, FetchRequest)> { + let plan = query_plan_from_ir_json(query_ir_json)?; validate_paging_shape(&plan, "fetch")?; validate_set_shape(&plan, "fetch")?; let include_paths: Option>> = match &plan.materialization { @@ -3376,6 +3379,207 @@ pub fn fetch_filtered<'py>( )); } }; + Ok(( + plan, + FetchRequest { + include_paths, + projected, + record_cls_py, + hop_classes_py, + }, + )) +} + +/// A fetch's statement, and what decoding its rows needs. +struct BuiltFetch { + statement: BuiltStatement, + /// The root model's registration: its codec plan and primary key. + schema: Arc, + join_plan: JoinPlan, + include_hops: Vec, +} + +/// Build the SELECT a fetch sends: the one build both +/// [`fetch_filtered`] (which sends it) and [`render_fetch`] (which returns +/// it) use. On Postgres it reads the root's and each hop table's cached +/// catalog (probing on first touch) so the binds carry the casts the +/// database needs. +/// +/// # Errors +/// `PyRuntimeError` for an unregistered model or a failed catalog probe; +/// `PyValueError` for a relation path, order key or paging bound the plan +/// cannot render. +async fn build_fetch( + plan: &mut QueryPlan, + name: &str, + include_paths: Option<&[Vec]>, + projected: Option<&[ProjectedField]>, + engine: &EngineHandle, + exec: Executor<'_>, + backend: Dialect, +) -> PyResult { + let schema = crate::state::registered_model(name)?; + let table_name = schema.table_name.clone(); + let catalog = postgres_table_catalog(&table_name, engine, exec, backend).await?; + plan.postgres_enum_udt = catalog.enum_udt.clone(); + + // Relation-traversal JOINs (#270): the M2M association table (when + // present) is reserved so a relation alias can never collide with it. + // The join plan is built up front so each hop table's registration + + // enum catalog can be resolved before conditions bind (#270 critical). + let m2m_join_table = plan.m2m.as_ref().map(|m2m| m2m.join_table.clone()); + let reserved: Vec<&str> = m2m_join_table.as_deref().into_iter().collect(); + let join_plan = query_join_plan(plan, &table_name, &reserved)?; + populate_hop_bind_context(plan, &join_plan, engine, exec, backend).await?; + // Included hops (#286): resolved after the bind context, so every + // include-edge table's registration is already in place. + let include_hops = match include_paths { + Some(paths) => resolve_include_hops(paths, &join_plan, plan)?, + None => Vec::new(), + }; + let statement = { + let pk = schema.meta.pk_col.clone(); + + let mut select = Query::select(); + apply_select_list( + &mut select, + &table_name, + &schema.column_names, + projected, + &join_plan, + )?; + apply_include_select_list(&mut select, &include_hops); + select.from(Alias::new(&table_name)); + + if let Some(m2m) = &plan.m2m { + let join_table = Alias::new(&m2m.join_table); + let source_col = Alias::new(&m2m.source_col); + let target_col = Alias::new(&m2m.target_col); + let pk_name = pk.as_ref().ok_or_else(|| { + pyo3::exceptions::PyRuntimeError::new_err("No primary key for M2M join") + })?; + + select.inner_join( + join_table.clone(), + Expr::col((Alias::new(&table_name), Alias::new(pk_name))) + .equals((join_table.clone(), target_col.clone())), + ); + select.and_where(Expr::col((join_table.clone(), source_col.clone())).eq( + plan.value_rhs_simple_expr_for_backend( + &m2m.source_col, + &m2m.source_id, + true, + backend, + ), + )); + } + + apply_relation_joins(&mut select, &join_plan)?; + + // WHERE columns are qualified by their relation-path alias (root leaf -> + // root table, path leaf -> its JOIN alias). A path with no matching join + // entry is a loud error, never a silently unqualified column. + let mut condition = query_condition_with_joins(plan, backend, &table_name, &join_plan)?; + if let Some(after) = plan + .after_condition(backend, &table_name, &join_plan) + .map_err(pyo3::exceptions::PyValueError::new_err)? + { + condition = condition.add(after); + } + if let Some(before) = plan + .before_condition(backend, &table_name, &join_plan) + .map_err(pyo3::exceptions::PyValueError::new_err)? + { + condition = condition.add(before); + } + select.cond_where(condition); + // ORDER BY terms are qualified the same way as WHERE leaves: an empty + // path qualifies by the root table, a relation path by its JOIN alias + // (#271). A path with no matching join entry is a loud error — the + // Python builder always registers the join for an order_by path, so + // this is defense-in-depth, never reachable in normal use. On a + // record plan the term resolves output field names FIRST (#295): + // a matching name renders as the bare result-column alias. + // `before` inverts direction and first↔last (`native` stays + // `native`) so LIMIT n fetches the adjacent previous page (#395). + let order_by = plan + .order_by_for_select() + .map_err(pyo3::exceptions::PyValueError::new_err)?; + for order in &order_by { + let col = match projected { + Some(fields) => record_order_by_expr(order, fields, &table_name, &join_plan)?, + None => crate::query::qualify_column_with_joins( + &table_name, + &join_plan, + &order.column, + &order.path, + ) + .map_err(pyo3::exceptions::PyValueError::new_err)?, + }; + apply_order_by_term(&mut select, col.into(), order)?; + } + if let Some(limit) = plan.limit.flatten() { + select.limit(limit); + } + if let Some(offset) = plan.offset.flatten() { + select.offset(offset); + } + let (sql, values) = sea_query_build_for_backend!(select, backend); + BuiltStatement::from_sea(sql, &values) + }; + + Ok(BuiltFetch { + statement, + schema, + join_plan, + include_hops, + }) +} + +/// Fetches records for a given model class based on a QueryIR-defined query. +/// +/// The query's materialization plan (ADR-0007) selects the result shape: +/// `root_instances` hydrates complete model instances (identity-map aware); +/// a `record` plan renders the projected SELECT list and hydrates +/// `record_cls` records (#279) — no identity map, no persistence identity; +/// an `instances` plan widens the SELECT with every included hop's complete +/// aliased row and hydrates a populated instance graph (#286, ADR-0008). +/// +/// Args: +/// cls (PyAny): The Python model class. +/// query_ir_json (str): The serialized QueryIR envelope JSON. +/// record_cls (PyAny | None): Record constructor for a `record` plan +/// (`Row` today; the seam a future `into=` plugs into). Required for +/// a record plan, forbidden otherwise. +/// hop_classes (dict[str, type] | None): Included hop model classes keyed +/// by physical table name, for an `instances` plan. Required for an +/// instances plan, forbidden otherwise. +/// +/// Returns: +/// list[PyAny]: Hydrated model instances (populated when included), or +/// projected records. +#[pyfunction] +#[pyo3(signature = ( + cls, query_ir_json, route, record_cls=None, hop_classes=None, *, secret_values=Vec::new() +))] +pub fn fetch_filtered<'py>( + py: Python<'py>, + cls: Bound<'py, PyAny>, + query_ir_json: String, + route: Py, + record_cls: Option>, + hop_classes: Option>, + secret_values: Vec>, +) -> PyResult> { + let name = crate::state::model_identity(&cls)?; + let (mut plan, request) = fetch_request(&query_ir_json, record_cls, hop_classes)?; + let FetchRequest { + include_paths, + projected, + record_cls_py, + hop_classes_py, + } = request; + let cls_py = cls.unbind(); let log = operation_log(py, "fetch", &route.get().connection_name, |py| { operation_secret_values( &secret_values, @@ -3400,124 +3604,24 @@ pub fn fetch_filtered<'py>( let outcome = async { let use_identity_map = engine.is_identity_map_enabled(); - - let schema = crate::state::registered_model(&name)?; - let table_name = schema.table_name.clone(); - let catalog = postgres_table_catalog(&table_name, &engine, exec, backend).await?; - plan.postgres_enum_udt = catalog.enum_udt.clone(); - - // Relation-traversal JOINs (#270): the M2M association table (when - // present) is reserved so a relation alias can never collide with it. - // The join plan is built up front so each hop table's registration + - // enum catalog can be resolved before conditions bind (#270 critical). - let m2m_join_table = plan.m2m.as_ref().map(|m2m| m2m.join_table.clone()); - let reserved: Vec<&str> = m2m_join_table.as_deref().into_iter().collect(); - let join_plan = query_join_plan(&plan, &table_name, &reserved)?; - populate_hop_bind_context(&mut plan, &join_plan, &engine, exec, backend).await?; - // Included hops (#286): resolved after the bind context, so every - // include-edge table's registration is already in place. - let include_hops = match &include_paths { - Some(paths) => resolve_include_hops(paths, &join_plan, &plan)?, - None => Vec::new(), - }; - // ... - let (sql, bind_values, pk_col, schema_for_decode) = { - let pk = schema.meta.pk_col.clone(); - - let mut select = Query::select(); - apply_select_list( - &mut select, - &table_name, - &schema.column_names, - projected.as_deref(), - &join_plan, - )?; - apply_include_select_list(&mut select, &include_hops); - select.from(Alias::new(&table_name)); - - if let Some(m2m) = &plan.m2m { - let join_table = Alias::new(&m2m.join_table); - let source_col = Alias::new(&m2m.source_col); - let target_col = Alias::new(&m2m.target_col); - let pk_name = pk.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("No primary key for M2M join") - })?; - - select.inner_join( - join_table.clone(), - Expr::col((Alias::new(&table_name), Alias::new(pk_name))) - .equals((join_table.clone(), target_col.clone())), - ); - select.and_where(Expr::col((join_table.clone(), source_col.clone())).eq( - plan.value_rhs_simple_expr_for_backend( - &m2m.source_col, - &m2m.source_id, - true, - backend, - ), - )); - } - - apply_relation_joins(&mut select, &join_plan)?; - - // WHERE columns are qualified by their relation-path alias (root leaf -> - // root table, path leaf -> its JOIN alias). A path with no matching join - // entry is a loud error, never a silently unqualified column. - let mut condition = - query_condition_with_joins(&plan, backend, &table_name, &join_plan)?; - if let Some(after) = plan - .after_condition(backend, &table_name, &join_plan) - .map_err(pyo3::exceptions::PyValueError::new_err)? - { - condition = condition.add(after); - } - if let Some(before) = plan - .before_condition(backend, &table_name, &join_plan) - .map_err(pyo3::exceptions::PyValueError::new_err)? - { - condition = condition.add(before); - } - select.cond_where(condition); - // ORDER BY terms are qualified the same way as WHERE leaves: an empty - // path qualifies by the root table, a relation path by its JOIN alias - // (#271). A path with no matching join entry is a loud error — the - // Python builder always registers the join for an order_by path, so - // this is defense-in-depth, never reachable in normal use. On a - // record plan the term resolves output field names FIRST (#295): - // a matching name renders as the bare result-column alias. - // `before` inverts direction and first↔last (`native` stays - // `native`) so LIMIT n fetches the adjacent previous page (#395). - let order_by = plan - .order_by_for_select() - .map_err(pyo3::exceptions::PyValueError::new_err)?; - for order in &order_by { - let col = match projected.as_deref() { - Some(fields) => { - record_order_by_expr(order, fields, &table_name, &join_plan)? - } - None => crate::query::qualify_column_with_joins( - &table_name, - &join_plan, - &order.column, - &order.path, - ) - .map_err(pyo3::exceptions::PyValueError::new_err)?, - }; - apply_order_by_term(&mut select, col.into(), order)?; - } - if let Some(limit) = plan.limit.flatten() { - select.limit(limit); - } - if let Some(offset) = plan.offset.flatten() { - select.offset(offset); - } - let (s, values) = sea_query_build_for_backend!(select, backend); - (s, values, pk, schema.clone()) - }; - - let engine_bind_values = engine_bind_values_from_sea(&bind_values.0); + let BuiltFetch { + statement, + schema: schema_for_decode, + join_plan, + include_hops, + } = build_fetch( + &mut plan, + &name, + include_paths.as_deref(), + projected.as_deref(), + &engine, + exec, + backend, + ) + .await?; + let pk_col = schema_for_decode.meta.pk_col.clone(); let rows = exec - .fetch_all(&sql, &engine_bind_values) + .fetch_all(&statement.sql, &statement.binds) .await .map_err(|e| crate::errors::map_db_error("Fetch failed", e))?; // A record plan skips PK extraction entirely: projected records carry @@ -3764,6 +3868,109 @@ pub fn fetch_filtered<'py>( pyo3_async_runtimes::tokio::future_into_py(py, crate::sql_log::scope(log, operation)) } +/// Parse and check a count's plan, and name its statement-log records: +/// `"count"`, or `"exists"` when the count answers `exists()`. +/// +/// # Errors +/// `PyValueError` for an unknown `operation` or a plan a count cannot run. +fn count_plan(query_ir_json: &str, operation: &str) -> PyResult<(QueryPlan, &'static str)> { + let operation: &'static str = match operation { + "count" => "count", + "exists" => "exists", + other => { + return Err(pyo3::exceptions::PyValueError::new_err(format!( + "count_filtered operation must be 'count' or 'exists', got {other:?}" + ))); + } + }; + let plan = query_plan_from_ir_json(query_ir_json)?; + validate_paging_shape(&plan, "count")?; + validate_set_shape(&plan, "count")?; + // count() is unaffected by projection (PRD #277 verb table): the Python + // builder always emits root_instances on count payloads, projected or not. + reject_unsupported_materialization(&plan, "count_filtered")?; + Ok((plan, operation)) +} + +/// Build the `SELECT COUNT(*)` that `count()` and `exists()` send: the +/// one build [`count_filtered`] and [`render_count`] share. On Postgres +/// it reads the root's and each hop table's cached catalog. +/// +/// # Errors +/// `PyRuntimeError` for an unregistered model or a failed catalog probe; +/// `PyTypeError` / `PyValueError` for a value or relation path the plan +/// cannot render. +async fn build_count( + plan: &mut QueryPlan, + name: &str, + engine: &EngineHandle, + exec: Executor<'_>, + backend: Dialect, +) -> PyResult { + let schema = crate::state::registered_model(name)?; + let table_name = schema.table_name.clone(); + plan.postgres_enum_udt = postgres_table_catalog(&table_name, engine, exec, backend) + .await? + .enum_udt + .clone(); + + // Relation-traversal JOINs render into the COUNT the same as the SELECT + // (#270). Plain COUNT(*), no DISTINCT: forward-FK hops are many-to-one and + // never multiply root rows, so count() equals the matching root-row count. + // Build the plan up front and resolve each hop table's registration + + // enum catalog before conditions bind (#270 critical). + let m2m_join_table = plan.m2m.as_ref().map(|m2m| m2m.join_table.clone()); + let reserved: Vec<&str> = m2m_join_table.as_deref().into_iter().collect(); + let join_plan = query_join_plan(plan, &table_name, &reserved)?; + populate_hop_bind_context(plan, &join_plan, engine, exec, backend).await?; + let (sql, values) = { + let mut select = Query::select(); + select.expr(Expr::cust("COUNT(*)")); + + if let Some(m2m) = &plan.m2m { + let join_table = Alias::new(&m2m.join_table); + let source_col = Alias::new(&m2m.source_col); + let target_col = Alias::new(&m2m.target_col); + + // We need the PK name of the target table to join + let pk_name = schema + .meta + .pk_col + .clone() + .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("No primary key"))?; + + select.from(Alias::new(&table_name)); + select.inner_join( + join_table.clone(), + Expr::col((Alias::new(&table_name), Alias::new(pk_name))) + .equals((join_table.clone(), target_col.clone())), + ); + select.and_where(Expr::col((join_table.clone(), source_col.clone())).eq( + plan.value_rhs_simple_expr_for_backend( + &m2m.source_col, + &m2m.source_id, + true, + backend, + ), + )); + } else { + select.from(Alias::new(&table_name)); + } + + apply_relation_joins(&mut select, &join_plan)?; + + // WHERE columns qualified by their relation-path alias (see fetch_filtered). + select.cond_where(query_condition_with_joins( + plan, + backend, + &table_name, + &join_plan, + )?); + sea_query_build_for_backend!(select, backend) + }; + Ok(BuiltStatement::from_sea(sql, &values)) +} + /// Return the number of rows matching a filtered query. /// /// Args: @@ -3793,21 +4000,7 @@ pub fn count_filtered<'py>( secret_values: Vec>, operation: &str, ) -> PyResult> { - let operation: &'static str = match operation { - "count" => "count", - "exists" => "exists", - other => { - return Err(pyo3::exceptions::PyValueError::new_err(format!( - "count_filtered operation must be 'count' or 'exists', got {other:?}" - ))); - } - }; - let mut plan = query_plan_from_ir_json(&query_ir_json)?; - validate_paging_shape(&plan, "count")?; - validate_set_shape(&plan, "count")?; - // count() is unaffected by projection (PRD #277 verb table): the Python - // builder always emits root_instances on count payloads, projected or not. - reject_unsupported_materialization(&plan, "count_filtered")?; + let (mut plan, operation) = count_plan(&query_ir_json, operation)?; let log = operation_log(py, operation, &route.get().connection_name, |py| { operation_secret_values( &secret_values, @@ -3830,70 +4023,9 @@ pub fn count_filtered<'py>( let exec = Executor::new(scope.connection(), &engine); let outcome = async { - let schema = crate::state::registered_model(&name)?; - let table_name = schema.table_name.clone(); - plan.postgres_enum_udt = postgres_table_catalog(&table_name, &engine, exec, backend) - .await? - .enum_udt - .clone(); - - // Relation-traversal JOINs render into the COUNT the same as the SELECT - // (#270). Plain COUNT(*), no DISTINCT: forward-FK hops are many-to-one and - // never multiply root rows, so count() equals the matching root-row count. - // Build the plan up front and resolve each hop table's registration + - // enum catalog before conditions bind (#270 critical). - let m2m_join_table = plan.m2m.as_ref().map(|m2m| m2m.join_table.clone()); - let reserved: Vec<&str> = m2m_join_table.as_deref().into_iter().collect(); - let join_plan = query_join_plan(&plan, &table_name, &reserved)?; - populate_hop_bind_context(&mut plan, &join_plan, &engine, exec, backend).await?; - // ... sql ... - let (sql, bind_values) = { - let mut select = Query::select(); - select.expr(Expr::cust("COUNT(*)")); - - if let Some(m2m) = &plan.m2m { - let join_table = Alias::new(&m2m.join_table); - let source_col = Alias::new(&m2m.source_col); - let target_col = Alias::new(&m2m.target_col); - - // We need the PK name of the target table to join - let pk_name = schema.meta.pk_col.clone().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("No primary key") - })?; - - select.from(Alias::new(&table_name)); - select.inner_join( - join_table.clone(), - Expr::col((Alias::new(&table_name), Alias::new(pk_name))) - .equals((join_table.clone(), target_col.clone())), - ); - select.and_where(Expr::col((join_table.clone(), source_col.clone())).eq( - plan.value_rhs_simple_expr_for_backend( - &m2m.source_col, - &m2m.source_id, - true, - backend, - ), - )); - } else { - select.from(Alias::new(&table_name)); - } - - apply_relation_joins(&mut select, &join_plan)?; - - // WHERE columns qualified by their relation-path alias (see fetch_filtered). - select.cond_where(query_condition_with_joins( - &plan, - backend, - &table_name, - &join_plan, - )?); - sea_query_build_for_backend!(select, backend) - }; - - let engine_bind_values = engine_bind_values_from_sea(&bind_values.0); + let statement = build_count(&mut plan, &name, &engine, exec, backend).await?; let rows = exec - .fetch_all(&sql, &engine_bind_values) + .fetch_all(&statement.sql, &statement.binds) .await .map_err(|e| crate::errors::map_db_error("Count failed", e))?; let count = rows @@ -4037,6 +4169,58 @@ pub fn delete_record( pyo3_async_runtimes::tokio::future_into_py(py, crate::sql_log::scope(log, operation)) } +/// Parse and check a query delete's plan. +/// +/// # Errors +/// `PyValueError` for a plan a delete cannot run: paging, a traversal, a +/// projection or a `SET`. +fn delete_plan(query_ir_json: &str) -> PyResult { + let plan = query_plan_from_ir_json(query_ir_json)?; + validate_set_shape(&plan, "delete")?; + validate_paging_shape(&plan, "delete")?; + reject_traversal_on_mutation(&plan, "delete")?; + reject_unsupported_materialization(&plan, "delete")?; + Ok(plan) +} + +/// Build the `DELETE` a query's `delete()` sends: the one build +/// [`delete_filtered`] and [`render_delete`] share. On Postgres it reads +/// the table's cached catalog. +/// +/// # Errors +/// `PyRuntimeError` for an unregistered model or a failed catalog probe; +/// `PyTypeError` / `PyValueError` for a value or relation path the plan +/// cannot render. +async fn build_delete( + plan: &mut QueryPlan, + name: &str, + engine: &EngineHandle, + exec: Executor<'_>, + backend: Dialect, +) -> PyResult { + let table_name = crate::state::registered_model(name)?.table_name.clone(); + plan.postgres_enum_udt = postgres_table_catalog(&table_name, engine, exec, backend) + .await? + .enum_udt + .clone(); + let (sql, values) = { + let mut delete = Query::delete(); + // Qualified by the root table alias (#269): both dialects accept + // `table.column` in `DELETE ... WHERE`, and uniform qualification across + // every walker avoids retrofitting the mutating paths when joins reach + // them later. + delete + .from_table(Alias::new(&table_name)) + .cond_where(query_condition_for_backend( + plan, + backend, + Some(&table_name), + )?); + sea_query_build_for_backend!(delete, backend) + }; + Ok(BuiltStatement::from_sea(sql, &values)) +} + /// Delete rows matching a filtered query. /// /// Args: @@ -4060,11 +4244,7 @@ pub fn delete_filtered<'py>( route: Py, secret_values: Vec>, ) -> PyResult> { - let mut plan = query_plan_from_ir_json(&query_ir_json)?; - validate_set_shape(&plan, "delete")?; - validate_paging_shape(&plan, "delete")?; - reject_traversal_on_mutation(&plan, "delete")?; - reject_unsupported_materialization(&plan, "delete")?; + let mut plan = delete_plan(&query_ir_json)?; let log = operation_log(py, "delete", &route.get().connection_name, |py| { operation_secret_values( &secret_values, @@ -4088,33 +4268,11 @@ pub fn delete_filtered<'py>( let exec = Executor::new(scope.connection(), &engine); let outcome = async { - let table_name = crate::state::registered_model(&name)?.table_name.clone(); - plan.postgres_enum_udt = postgres_table_catalog(&table_name, &engine, exec, backend) - .await? - .enum_udt - .clone(); - // ... sql ... - let (sql, bind_values) = - { - let mut delete = Query::delete(); - // Qualified by the root table alias (#269): both dialects accept - // `table.column` in `DELETE ... WHERE`, and uniform qualification across - // every walker avoids retrofitting the mutating paths when joins reach - // them later. - delete.from_table(Alias::new(&table_name)).cond_where( - query_condition_for_backend(&plan, backend, Some(&table_name))?, - ); - sea_query_build_for_backend!(delete, backend) - }; - - let rows_affected = execute_statement_with_optional_tx( - &engine, - scope.connection(), - &sql, - &bind_values.0, - ) - .await - .map_err(|e| crate::errors::map_db_error("Delete failed", e))?; + let statement = build_delete(&mut plan, &name, &engine, exec, backend).await?; + let rows_affected = exec + .execute(&statement.sql, &statement.binds) + .await + .map_err(|e| crate::errors::map_db_error("Delete failed", e))?; // After bulk delete, we MUST clear the Identity Map for this model to avoid stale objects // Re-acquire the GIL before accessing identity map (async context has no GIL). @@ -4133,6 +4291,85 @@ pub fn delete_filtered<'py>( pyo3_async_runtimes::tokio::future_into_py(py, crate::sql_log::scope(log, operation)) } +/// Parse and check a query update's plan against the model it names. +/// +/// # Errors +/// `PyValueError` for a plan an update cannot run (no `SET`, paging, a +/// traversal, a projection) or one declaring another model than `name`. +fn update_plan(name: &str, query_ir_json: &str) -> PyResult { + let plan = query_plan_from_ir_json(query_ir_json)?; + validate_set_shape(&plan, "update")?; + if name != plan.model_name { + return Err(pyo3::exceptions::PyValueError::new_err(format!( + "update model identity mismatch: argument {name:?}, QueryIR declares {:?}", + plan.model_name + ))); + } + validate_paging_shape(&plan, "update")?; + reject_traversal_on_mutation(&plan, "update")?; + reject_unsupported_materialization(&plan, "update")?; + Ok(plan) +} + +/// Build the `UPDATE` a query's `update()` sends, from either form (the +/// keyword literals or the recipe, both already in the plan's `SET`): +/// the one build [`update_filtered`] and [`render_update`] share. On +/// Postgres it reads the table's cached catalog, whose column types +/// decide the `SET` binds' casts. +/// +/// # Errors +/// `PyRuntimeError` for an unregistered model or a failed catalog probe; +/// `PyTypeError` / `PyValueError` for a value or relation path the plan +/// cannot render. +async fn build_update( + plan: &mut QueryPlan, + name: &str, + engine: &EngineHandle, + exec: Executor<'_>, + backend: Dialect, +) -> PyResult { + let schema = crate::state::registered_model(name)?; + let table_name = schema.table_name.clone(); + let catalog = postgres_table_catalog(&table_name, engine, exec, backend).await?; + let TableCatalog { + enum_udt, + uuid_columns, + ts_cast, + } = &*catalog; + plan.postgres_enum_udt = enum_udt.clone(); + let (sql, values) = { + // Qualified by the root table alias (#269) — same rationale as + // `delete_filtered` above. SET columns stay bare: SQL forbids + // qualified column names on the left of `SET` assignments. + let mut update = UpdateStatement::new() + .table(Alias::new(&table_name)) + .cond_where(query_condition_for_backend( + plan, + backend, + Some(&table_name), + )?) + .to_owned(); + for assignment in &plan.set_assignments { + let key = &assignment.column; + update.value( + Alias::new(key), + set_value_to_expr( + &schema, + &table_name, + key, + &assignment.value, + enum_udt, + uuid_columns, + ts_cast, + backend, + )?, + ); + } + sea_query_build_for_backend!(update, backend) + }; + Ok(BuiltStatement::from_sea(sql, &values)) +} + /// Update rows matching a filtered query using its canonical QueryIR SET section. /// /// Args: @@ -4156,17 +4393,7 @@ pub fn update_filtered<'py>( route: Py, secret_values: Vec>, ) -> PyResult> { - let mut plan = query_plan_from_ir_json(&query_ir_json)?; - validate_set_shape(&plan, "update")?; - if name != plan.model_name { - return Err(pyo3::exceptions::PyValueError::new_err(format!( - "update model identity mismatch: argument {name:?}, QueryIR declares {:?}", - plan.model_name - ))); - } - validate_paging_shape(&plan, "update")?; - reject_traversal_on_mutation(&plan, "update")?; - reject_unsupported_materialization(&plan, "update")?; + let mut plan = update_plan(&name, &query_ir_json)?; let log = operation_log(py, "update", &route.get().connection_name, |py| { operation_secret_values( &secret_values, @@ -4189,55 +4416,11 @@ pub fn update_filtered<'py>( let exec = Executor::new(scope.connection(), &engine); let outcome = async { - let schema = crate::state::registered_model(&name)?; - let table_name = schema.table_name.clone(); - let catalog = postgres_table_catalog(&table_name, &engine, exec, backend).await?; - let TableCatalog { - enum_udt, - uuid_columns, - ts_cast, - } = &*catalog; - plan.postgres_enum_udt = enum_udt.clone(); - // ... sql ... - let (sql, bind_values) = { - // Qualified by the root table alias (#269) — same rationale as - // `delete_filtered` above. SET columns stay bare: SQL forbids - // qualified column names on the left of `SET` assignments. - let mut update = UpdateStatement::new() - .table(Alias::new(&table_name)) - .cond_where(query_condition_for_backend( - &plan, - backend, - Some(&table_name), - )?) - .to_owned(); - for assignment in &plan.set_assignments { - let key = &assignment.column; - update.value( - Alias::new(key), - set_value_to_expr( - &schema, - &table_name, - key, - &assignment.value, - enum_udt, - uuid_columns, - ts_cast, - backend, - )?, - ); - } - sea_query_build_for_backend!(update, backend) - }; - - let rows_affected = execute_statement_with_optional_tx( - &engine, - scope.connection(), - &sql, - &bind_values.0, - ) - .await - .map_err(|e| crate::errors::map_db_error("Update failed", e))?; + let statement = build_update(&mut plan, &name, &engine, exec, backend).await?; + let rows_affected = exec + .execute(&statement.sql, &statement.binds) + .await + .map_err(|e| crate::errors::map_db_error("Update failed", e))?; // After bulk update, we MUST clear the Identity Map for this model to avoid stale objects // Re-acquire the GIL before accessing identity map (async context has no GIL). @@ -4256,6 +4439,226 @@ pub fn update_filtered<'py>( pyo3_async_runtimes::tokio::future_into_py(py, crate::sql_log::scope(log, operation)) } +/// A rendering entry's answer: the statement's SQL, its parameters as the +/// statement log shows them (`None` where masked), the mask and the +/// dialect's name. +/// +/// # Errors +/// The `PyErr` from converting a bind to Python. +fn rendered_parts( + statement: &BuiltStatement, + secrets: &[EngineBindValue], + backend: Dialect, +) -> PyResult> { + use pyo3::IntoPyObjectExt; + Python::attach(|py| { + let (params, mask) = crate::sql_log::shown_params(py, &statement.binds, secrets)?; + Ok(pyo3::types::PyTuple::new( + py, + [ + statement.sql.as_str().into_bound_py_any(py)?, + params.into_any(), + mask.into_any(), + crate::sql_log::dialect_name(backend).into_bound_py_any(py)?, + ], + )? + .into_any() + .unbind()) + }) +} + +/// Render one query statement for `route` without sending it: resolve the +/// route's engine, take its connection the way the call would but open no +/// transaction of its own (a session's pinned connection, or the open +/// `transaction()`'s, still serves the catalog probe), and run `build`. +/// Only a first-touch catalog probe reaches the database. +macro_rules! render_on_route { + ($route:expr, $secrets:expr, |$engine:ident, $exec:ident, $backend:ident| $build:expr) => {{ + let r = $route.get(); + let (_, engine, tx_conn, $backend) = route_engine(r)?; + let scope = OperationScope::unwrapped(&engine, r.session_id.as_deref(), tx_conn).await?; + let built = { + let $engine: &EngineHandle = &engine; + let $exec = Executor::new(scope.connection(), &engine); + $build.await + }; + let statement = scope.finish(built).await?; + rendered_parts(&statement, &$secrets, $backend) + }}; +} + +/// Render the `SELECT` [`fetch_filtered`] would send for the same +/// arguments, without sending it (`all()` / `first()` `.sql()`). +/// +/// Args: +/// cls (PyAny): The Python model class. +/// query_ir_json (str): The serialized QueryIR envelope JSON. +/// route (RouteHandle): The resolved route. +/// record_cls (PyAny | None): As for `fetch_filtered`. +/// hop_classes (dict[str, type] | None): As for `fetch_filtered`. +/// secret_values (list[object]): The secret-typed literals the query +/// compiled with, masked in the answer. +/// +/// Returns: +/// tuple[str, tuple, tuple[bool, ...], str]: The SQL, the parameters +/// (`None` where masked), the mask and the dialect. +/// +/// # Errors +/// The errors `fetch_filtered` raises before it sends. +#[pyfunction] +#[pyo3(signature = ( + cls, query_ir_json, route, record_cls=None, hop_classes=None, *, secret_values=Vec::new() +))] +pub fn render_fetch<'py>( + py: Python<'py>, + cls: Bound<'py, PyAny>, + query_ir_json: String, + route: Py, + record_cls: Option>, + hop_classes: Option>, + secret_values: Vec>, +) -> PyResult> { + let name = crate::state::model_identity(&cls)?; + let (mut plan, request) = fetch_request(&query_ir_json, record_cls, hop_classes)?; + let secrets = operation_secret_values(&secret_values, plan_secret_values(py, &cls, &plan)?)?; + let log = operation_log(py, "fetch", &route.get().connection_name, no_secrets)?; + let rendering = async move { + render_on_route!(route, secrets, |engine, exec, backend| async { + build_fetch( + &mut plan, + &name, + request.include_paths.as_deref(), + request.projected.as_deref(), + engine, + exec, + backend, + ) + .await + .map(|built| built.statement) + }) + }; + pyo3_async_runtimes::tokio::future_into_py(py, crate::sql_log::scope(log, rendering)) +} + +/// Render the `SELECT COUNT(*)` [`count_filtered`] would send for the same +/// arguments, without sending it (`count()` / `exists()` `.sql()`). +/// +/// Args: +/// name (str): Model class name. +/// query_ir_json (str): Serialized Query IR envelope JSON. +/// route (RouteHandle): The resolved route. +/// secret_values (list[object]): The secret-typed literals the query +/// compiled with, masked in the answer. +/// operation (str): `"count"` or `"exists"`, as for `count_filtered`. +/// +/// Returns: +/// tuple[str, tuple, tuple[bool, ...], str]: The SQL, the parameters +/// (`None` where masked), the mask and the dialect. +/// +/// # Errors +/// The errors `count_filtered` raises before it sends. +#[pyfunction] +#[pyo3(signature = (name, query_ir_json, route, *, secret_values=Vec::new(), operation="count"))] +pub fn render_count<'py>( + py: Python<'py>, + name: String, + query_ir_json: String, + route: Py, + secret_values: Vec>, + operation: &str, +) -> PyResult> { + let (mut plan, operation) = count_plan(&query_ir_json, operation)?; + let secrets = operation_secret_values( + &secret_values, + plan_secret_values(py, &model_class_for(py, &name)?, &plan)?, + )?; + let log = operation_log(py, operation, &route.get().connection_name, no_secrets)?; + let rendering = async move { + render_on_route!(route, secrets, |engine, exec, backend| build_count( + &mut plan, &name, engine, exec, backend + )) + }; + pyo3_async_runtimes::tokio::future_into_py(py, crate::sql_log::scope(log, rendering)) +} + +/// Render the `UPDATE` [`update_filtered`] would send for the same +/// arguments, without sending it (`update()` `.sql()`). +/// +/// Args: +/// name (str): Model class name. +/// query_ir_json (str): Serialized Query IR envelope JSON. +/// route (RouteHandle): The resolved route. +/// secret_values (list[object]): The secret-typed literals the query +/// compiled with, masked in the answer. +/// +/// Returns: +/// tuple[str, tuple, tuple[bool, ...], str]: The SQL, the parameters +/// (`None` where masked), the mask and the dialect. +/// +/// # Errors +/// The errors `update_filtered` raises before it sends. +#[pyfunction] +#[pyo3(signature = (name, query_ir_json, route, *, secret_values=Vec::new()))] +pub fn render_update<'py>( + py: Python<'py>, + name: String, + query_ir_json: String, + route: Py, + secret_values: Vec>, +) -> PyResult> { + let mut plan = update_plan(&name, &query_ir_json)?; + let secrets = operation_secret_values( + &secret_values, + plan_secret_values(py, &model_class_for(py, &name)?, &plan)?, + )?; + let log = operation_log(py, "update", &route.get().connection_name, no_secrets)?; + let rendering = async move { + render_on_route!(route, secrets, |engine, exec, backend| build_update( + &mut plan, &name, engine, exec, backend + )) + }; + pyo3_async_runtimes::tokio::future_into_py(py, crate::sql_log::scope(log, rendering)) +} + +/// Render the `DELETE` [`delete_filtered`] would send for the same +/// arguments, without sending it (`delete()` `.sql()`). +/// +/// Args: +/// name (str): Model class name. +/// query_ir_json (str): Serialized Query IR envelope JSON. +/// route (RouteHandle): The resolved route. +/// secret_values (list[object]): The secret-typed literals the query +/// compiled with, masked in the answer. +/// +/// Returns: +/// tuple[str, tuple, tuple[bool, ...], str]: The SQL, the parameters +/// (`None` where masked), the mask and the dialect. +/// +/// # Errors +/// The errors `delete_filtered` raises before it sends. +#[pyfunction] +#[pyo3(signature = (name, query_ir_json, route, *, secret_values=Vec::new()))] +pub fn render_delete<'py>( + py: Python<'py>, + name: String, + query_ir_json: String, + route: Py, + secret_values: Vec>, +) -> PyResult> { + let mut plan = delete_plan(&query_ir_json)?; + let secrets = operation_secret_values( + &secret_values, + plan_secret_values(py, &model_class_for(py, &name)?, &plan)?, + )?; + let log = operation_log(py, "delete", &route.get().connection_name, no_secrets)?; + let rendering = async move { + render_on_route!(route, secrets, |engine, exec, backend| build_delete( + &mut plan, &name, engine, exec, backend + )) + }; + pyo3_async_runtimes::tokio::future_into_py(py, crate::sql_log::scope(log, rendering)) +} + /// Insert many-to-many association rows into a join table. /// /// Args: diff --git a/src/sql_log.rs b/src/sql_log.rs index c38f41da..3c8cde6d 100644 --- a/src/sql_log.rs +++ b/src/sql_log.rs @@ -278,7 +278,35 @@ fn mask_of(binds: &[EngineBindValue], secrets: &[EngineBindValue]) -> Vec binds.iter().map(|bind| secrets.contains(bind)).collect() } -fn dialect_name(dialect: Dialect) -> &'static str { +/// A statement's parameters as a record shows them: each bind as its +/// Python value, `None` where it equals one of `secrets`, beside the mask +/// that marks those positions. The statement log and a rendered statement +/// (`.sql()`) both show parameters through this one function. +/// +/// # Errors +/// The `PyErr` from converting a bind to Python. +pub fn shown_params<'py>( + py: Python<'py>, + binds: &[EngineBindValue], + secrets: &[EngineBindValue], +) -> PyResult<(Bound<'py, PyTuple>, Bound<'py, PyTuple>)> { + let mask = mask_of(binds, secrets); + let params = binds + .iter() + .zip(&mask) + .map(|(bind, &masked)| { + if masked { + Ok(py.None().into_bound(py)) + } else { + bind_to_py(py, bind) + } + }) + .collect::>>()?; + Ok((PyTuple::new(py, params)?, PyTuple::new(py, mask)?)) +} + +/// The name a record gives `dialect`: `"postgres"` or `"sqlite"`. +pub fn dialect_name(dialect: Dialect) -> &'static str { match dialect { Dialect::Postgres => "postgres", Dialect::Sqlite => "sqlite", @@ -314,26 +342,15 @@ fn emit( duration_ms: f64, outcome: Result, &sqlx::Error>, ) -> PyResult<()> { - let mask = mask_of(binds, &context.secret_values); - let params = binds - .iter() - .zip(&mask) - .map(|(bind, &masked)| { - if masked { - Ok(py.None().into_bound(py)) - } else { - bind_to_py(py, bind) - } - }) - .collect::>>()?; + let (params, mask) = shown_params(py, binds, &context.secret_values)?; let (rows, error) = match outcome { Ok(rows) => (rows, None), Err(err) => (None, Some(err.to_string())), }; let record = PyDict::new(py); record.set_item("sql", sql)?; - record.set_item("params", PyTuple::new(py, params)?)?; - record.set_item("mask", PyTuple::new(py, mask)?)?; + record.set_item("params", params)?; + record.set_item("mask", mask)?; record.set_item("dialect", dialect_name(dialect))?; record.set_item("duration_ms", duration_ms)?; record.set_item("operation", context.operation)?; @@ -368,6 +385,23 @@ mod tests { assert_eq!(mask_of(&binds, &[]), vec![false, false, false]); } + #[test] + fn shown_params_puts_none_where_a_bind_is_secret() { + let secret = EngineBindValue::String("hunter2".into()); + let shown = Python::attach(|py| -> PyResult<(String, String)> { + let (params, mask) = + shown_params(py, &[EngineBindValue::I64(7), secret.clone()], &[secret])?; + Ok((params.repr()?.to_string(), mask.repr()?.to_string())) + }); + match shown { + Ok((params, mask)) => { + assert_eq!(params, "(7, None)"); + assert_eq!(mask, "(False, True)"); + } + Err(err) => panic!("shown_params failed: {err}"), + } + } + #[tokio::test] async fn outside_a_scope_there_is_no_context_and_observe_does_nothing() { assert!(current().is_none()); diff --git a/tests/static_fixtures/bad_updates.py b/tests/static_fixtures/bad_updates.py new file mode 100644 index 00000000..a37f25c1 --- /dev/null +++ b/tests/static_fixtures/bad_updates.py @@ -0,0 +1,26 @@ +"""Type-checked (never executed): mixing `update()`'s two forms must FAIL `ty check`. + +`update()` takes a recipe callable or keyword literals, never both, and so +does `update.sql()`, which takes the terminal's own arguments. +test_static_contracts.py asserts this file produces one `no-matching-overload` +diagnostic per misuse. If it starts passing, the overloads have stopped +biting. +""" + +from typing import Annotated + +from ferro import FerroField, Model +from ferro.query import Query + + +class BadUAccount(Model): + id: Annotated[int | None, FerroField(primary_key=True)] = None + visits: int = 0 + + +async def _mixed_forms_on_the_call() -> None: + await Query(BadUAccount).update(lambda account: {"visits": 1}, visits=1) + + +async def _mixed_forms_on_the_rendering() -> None: + await Query(BadUAccount).update.sql(lambda account: {"visits": 1}, visits=1) diff --git a/tests/static_fixtures/good_updates.py b/tests/static_fixtures/good_updates.py new file mode 100644 index 00000000..29a89853 --- /dev/null +++ b/tests/static_fixtures/good_updates.py @@ -0,0 +1,26 @@ +"""Type-checked (never executed): each of `update()`'s forms must PASS `ty check`. + +The recipe form and the keyword form type-check on the call and on +`update.sql()`, with the call's and the rendering's own return types. +""" + +from typing import Annotated, assert_type + +from ferro import FerroField, Model, RenderedSql +from ferro.query import Query + + +class GoodUAccount(Model): + id: Annotated[int | None, FerroField(primary_key=True)] = None + visits: int = 0 + + +async def _each_form() -> None: + query = Query(GoodUAccount).where(lambda account: account.id == 1) + assert_type(await query.update(visits=2), int) + assert_type(await query.update(lambda account: {"visits": account.visits + 1}), int) + assert_type(await query.update.sql(visits=2), RenderedSql) + assert_type( + await query.update.sql(lambda account: {"visits": account.visits + 1}), + RenderedSql, + ) diff --git a/tests/test_query_typing.py b/tests/test_query_typing.py index 63c73b1a..0866e483 100644 --- a/tests/test_query_typing.py +++ b/tests/test_query_typing.py @@ -170,3 +170,46 @@ class _StaticUser(Model): # FieldProxy comparisons resolve to QueryNode (the lambda mechanism) assert_type(FieldProxy("archived") == False, QueryNode) # noqa: E712 + + # Every terminal's `.sql()` renders with the terminal's own arguments and + # returns a RenderedSql, through the Relation and ProjectedQuery overrides + # too, while the call keeps its own return shape. + from ferro import RenderedSql, Row, Rows + + class _StaticAuthor(Model): + id: Annotated[int | None, FerroField(primary_key=True)] = None + posts: Relation[list["_StaticPost"]] = BackRef() + + class _StaticPost(Model): + id: Annotated[int | None, FerroField(primary_key=True)] = None + author: Annotated[_StaticAuthor, ForeignKey(related_name="posts")] + + async def _rendered_statement_shapes(author: _StaticAuthor) -> None: + query = Query(_StaticUser).where(lambda user: user.archived == False) # noqa: E712 + assert_type(await query.all(), list[_StaticUser]) + assert_type(await query.all.sql(), RenderedSql) + assert_type(await query.first(), _StaticUser | None) + assert_type(await query.first.sql(), RenderedSql) + assert_type(await query.count(), int) + assert_type(await query.count.sql(), RenderedSql) + assert_type(await query.exists.sql(), RenderedSql) + assert_type(await query.update(email="x"), int) + assert_type(await query.update.sql(email="x"), RenderedSql) + assert_type( + await query.update.sql(lambda user: {"email": user.email}), RenderedSql + ) + assert_type(await query.delete.sql(), RenderedSql) + + assert_type(await author.posts.all(), list[_StaticPost]) + assert_type(await author.posts.all.sql(), RenderedSql) + assert_type(await author.posts.first(), _StaticPost | None) + assert_type(await author.posts.first.sql(), RenderedSql) + assert_type(await author.posts.count.sql(), RenderedSql) + + projected = _StaticUser.select(lambda user: user.email) + assert_type(await projected.all(), Rows[Row]) + assert_type(await projected.all.sql(), RenderedSql) + assert_type(await projected.first(), Row | None) + assert_type(await projected.first.sql(), RenderedSql) + assert_type(await projected.count(), int) + assert_type(await projected.count.sql(), RenderedSql) diff --git a/tests/test_rendered_statement.py b/tests/test_rendered_statement.py new file mode 100644 index 00000000..58c25d26 --- /dev/null +++ b/tests/test_rendered_statement.py @@ -0,0 +1,369 @@ +"""The rendered statement: every `Query` terminal's `.sql()` (ADR-0057). + +```python +query = RsAccount.where(lambda account: account.visits >= 18) +rendered = await query.all.sql() +rendered.sql # 'SELECT ... WHERE "rsaccount"."visits" >= $1' +rendered.params # (18,) +rendered.dialect # 'postgres' +``` + +`.sql()` returns what the call would send without sending it: the same SQL +and parameters the statement log records for the call, built by the same +engine function, for the database the call would run on, with secrets shown +as `REDACTED`. The parity pin below holds that for every terminal on both +dialects: the rendered statement equals the `ferro.sql` record the executed +call writes. +""" + +import logging +from collections.abc import Awaitable, Callable +from typing import Annotated, Any + +import pytest +from pydantic import SecretStr + +import ferro +from ferro import ( + REDACTED, + BackRef, + Model, + Relation, + RenderedSql, + engines, +) +from ferro._core import identity_map_len +from ferro.base import FerroField, ForeignKey + +pytestmark = pytest.mark.backend_matrix + + +class RsAccount(Model): + id: Annotated[int | None, FerroField(primary_key=True)] = None + email: Annotated[str, FerroField(unique=True)] + token: SecretStr + visits: int = 0 + notes: Relation[list["RsNote"]] = BackRef() + + +class RsNote(Model): + id: Annotated[int | None, FerroField(primary_key=True)] = None + label: str + account: Annotated[RsAccount | None, ForeignKey(related_name="notes")] = None + + +async def _seeded(db_url: str) -> RsAccount: + """Connect, and save one account with one note.""" + await ferro.connect(db_url, auto_migrate=True) + async with engines.session(): + ada = RsAccount( + id=1, email="ada@example.com", token=SecretStr("sk_live_1"), visits=3 + ) + await ada.save() + await RsNote(id=1, label="hello", account=ada).save() + return ada + + +def _sent(records: list[logging.LogRecord]) -> list[logging.LogRecord]: + """The records of the statements the call itself sent: everything but the + pass `connect` runs and the catalog probe of a table's first touch.""" + return [r for r in records if r.operation not in ("migrate", "catalog")] + + +def _sentinel_positions(params: tuple[object, ...]) -> list[int]: + return [i for i, value in enumerate(params) if value is REDACTED] + + +# Each case: (name, the terminal to render and call, its arguments, whether a +# parameter is a secret). The terminal is read from a fresh query each time. +Terminal = Callable[[RsAccount], Any] +CASES: list[tuple[str, Terminal, tuple[Any, ...], dict[str, Any], bool]] = [ + ( + "all-secret-predicate", + lambda ada: RsAccount.where(lambda account: account.token == "sk_live_1").all, + (), + {}, + True, + ), + ( + "all-include", + lambda ada: ( + RsNote.select() + .include(lambda note: note.account) + .where(lambda note: note.label == "hello") + .order_by(lambda note: note.id) + .all + ), + (), + {}, + False, + ), + ( + "first", + lambda ada: ( + RsAccount.where(lambda account: account.visits >= 1) + .order_by(lambda account: account.id) + .first + ), + (), + {}, + False, + ), + ( + "count-traversed-secret", + lambda ada: RsNote.where(lambda note: note.account.token == "sk_live_1").count, + (), + {}, + True, + ), + ( + "exists", + lambda ada: ( + RsAccount.where(lambda account: account.email == "ada@example.com").exists + ), + (), + {}, + False, + ), + ( + "update-fields-secret", + lambda ada: RsAccount.where(lambda account: account.id == 1).update, + (), + {"token": SecretStr("sk_live_2")}, + True, + ), + ( + "update-recipe", + lambda ada: RsAccount.where(lambda account: account.id == 1).update, + (lambda account: {"visits": account.visits + 1},), + {}, + False, + ), + ( + "delete-secret", + lambda ada: ( + RsAccount.where(lambda account: account.token == "sk_live_1").delete + ), + (), + {}, + True, + ), + ( + "projected-all", + lambda ada: ( + RsAccount.select(lambda account: account.email) + .where(lambda account: account.visits >= 1) + .all + ), + (), + {}, + False, + ), + ( + "projected-count", + lambda ada: ( + RsAccount.select(lambda account: account.email) + .where(lambda account: account.token == "sk_live_1") + .count + ), + (), + {}, + True, + ), + ("relation-all", lambda ada: ada.notes.all, (), {}, False), + ("relation-first", lambda ada: ada.notes.first, (), {}, False), +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("terminal", "args", "kwargs", "secret"), + [case[1:] for case in CASES], + ids=[case[0] for case in CASES], +) +async def test_the_rendered_statement_is_the_logged_statement_of_the_call( + db_url, db_backend, statement_records, terminal, args, kwargs, secret +): + """The parity pin: `.sql()` equals the call's `ferro.sql` record in + `sql`, `params` (the `REDACTED` sentinel included) and `dialect`.""" + ada = await _seeded(db_url) + async with engines.session(): + rendered = await terminal(ada).sql(*args, **kwargs) + statement_records.clear() + await terminal(ada)(*args, **kwargs) + + (record,) = _sent(statement_records) + assert isinstance(rendered, RenderedSql) + assert rendered.sql == record.sql + assert rendered.params == record.params + assert _sentinel_positions(rendered.params) == _sentinel_positions(record.params) + assert rendered.dialect == record.dialect == db_backend + assert bool(_sentinel_positions(rendered.params)) is secret + assert "sk_live" not in repr(rendered.params) + + +@pytest.mark.asyncio +async def test_a_rendered_statement_prints_as_its_sql(db_url, db_backend): + await _seeded(db_url) + async with engines.session(): + rendered = await RsAccount.where(lambda account: account.visits >= 18).all.sql() + + placeholder = "$1" if db_backend == "postgres" else "?" + assert str(rendered) == rendered.sql + assert rendered.sql.startswith("SELECT ") + assert f'"rsaccount"."visits" >= {placeholder}' in rendered.sql + assert rendered.params == (18,) + assert rendered.dialect == db_backend + + +def test_a_rendered_statement_is_frozen(): + rendered = RenderedSql(sql="SELECT 1", params=(), dialect="sqlite") + with pytest.raises(AttributeError): + rendered.sql = "SELECT 2" # type: ignore[misc] # ty: ignore[invalid-assignment] + + +@pytest.mark.asyncio +async def test_update_sql_takes_both_forms_and_renders_them_differently(db_url): + await _seeded(db_url) + query = RsAccount.where(lambda account: account.id == 1) + async with engines.session(): + fields = await query.update.sql(visits=19) + recipe = await query.update.sql(lambda account: {"visits": account.visits + 1}) + + assert fields.sql.startswith("UPDATE ") + assert fields.params == (19, 1) + assert recipe.sql.startswith("UPDATE ") + assert recipe.params != fields.params + + +@pytest.mark.asyncio +async def test_first_sql_renders_the_limit_one_statement(db_url): + await _seeded(db_url) + async with engines.session(): + first = await RsAccount.where(lambda account: account.id == 1).first.sql() + every = await RsAccount.where(lambda account: account.id == 1).all.sql() + + assert "LIMIT" in first.sql + assert "LIMIT" not in every.sql + + +@pytest.mark.asyncio +@pytest.mark.parametrize("verb", ["update", "delete"]) +async def test_rendering_a_mutation_changes_no_row_and_sends_only_a_catalog_probe( + db_url, db_backend, statement_records, verb +): + await _seeded(db_url) + statement_records.clear() + async with engines.session() as session: + query = RsAccount.where(lambda account: account.id == 1) + if verb == "update": + await query.update.sql(visits=99) + else: + await query.delete.sql() + sent = list(statement_records) + assert identity_map_len(session.session_id) == 0 + statement_records.clear() + rows = await RsAccount.where(lambda account: account.id == 1).all() + + # Seeding already touched the table, so its catalog is cached. + assert sent == [] + assert [(row.id, row.visits) for row in rows] == [(1, 3)] + + +@pytest.mark.asyncio +@pytest.mark.postgres_only +async def test_rendering_on_a_tables_first_touch_sends_only_the_catalog_probe( + db_url, statement_records +): + """The casts in a rendered statement are the ones the call would use, so + on a table's first touch rendering reads its catalog: one `catalog` + record, and no record of the statement itself.""" + await ferro.connect(db_url, auto_migrate=True) + statement_records.clear() + async with engines.session(): + rendered = await RsAccount.where(lambda account: account.id == 1).delete.sql() + first = [record.operation for record in statement_records] + statement_records.clear() + again = await RsAccount.where(lambda account: account.id == 1).delete.sql() + + assert first == ["catalog"] + assert statement_records == [] + assert again == rendered + + +@pytest.mark.asyncio +async def test_rendering_a_fetch_puts_nothing_in_the_identity_map(db_url): + await _seeded(db_url) + async with engines.session() as session: + await RsAccount.where(lambda account: account.id == 1).all.sql() + await RsAccount.where(lambda account: account.id == 1).first.sql() + assert identity_map_len(session.session_id) == 0 + + +TERMINALS: list[tuple[str, Callable[[], Awaitable[Any]]]] = [ + ("all", lambda: RsAccount.where(lambda account: account.id == 1).all.sql()), + ("first", lambda: RsAccount.where(lambda account: account.id == 1).first.sql()), + ("count", lambda: RsAccount.where(lambda account: account.id == 1).count.sql()), + ("exists", lambda: RsAccount.where(lambda account: account.id == 1).exists.sql()), + ( + "update", + lambda: RsAccount.where(lambda account: account.id == 1).update.sql(visits=2), + ), + ("delete", lambda: RsAccount.where(lambda account: account.id == 1).delete.sql()), +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "render", [t[1] for t in TERMINALS], ids=[t[0] for t in TERMINALS] +) +async def test_rendering_outside_a_session_raises_the_no_route_error(db_url, render): + await _seeded(db_url) + with pytest.raises(RuntimeError, match="No database route for this operation"): + await render() + + +@pytest.mark.asyncio +async def test_update_sql_refuses_mixed_forms_like_update(db_url): + await _seeded(db_url) + query = RsAccount.where(lambda account: account.id == 1) + async with engines.session(): + with pytest.raises( + TypeError, match="either a recipe callable or keyword" + ) as call: + await query.update(lambda account: {"visits": 1}, visits=1) + with pytest.raises( + TypeError, match="either a recipe callable or keyword" + ) as render: + await query.update.sql(lambda account: {"visits": 1}, visits=1) + assert str(call.value) == str(render.value) + + +@pytest.mark.asyncio +@pytest.mark.postgres_only +async def test_using_and_session_route_the_rendering_to_their_database( + db_url, tmp_path +): + """`using=` and `session=` pick the database: the default is SQLite, + `pg` is Postgres, and each rendering speaks its route's dialect.""" + await ferro.connect(f"sqlite:{tmp_path / 'app.db'}?mode=rwc", default=True) + await ferro.create_tables() + await ferro.connect(db_url, name="pg", auto_migrate=True) + + async with engines.session(): + local = await RsAccount.where(lambda account: account.id == 1).all.sql() + async with engines.session("pg"): + by_using = await ( + RsAccount.using("pg").where(lambda account: account.id == 1).all.sql() + ) + async with engines.session("pg") as pg: + async with engines.session(): + by_session = await RsAccount.where( + lambda account: account.id == 1, session=pg + ).count.sql() + + assert (local.dialect, local.params) == ("sqlite", (1,)) + assert "?" in local.sql + assert (by_using.dialect, by_using.params) == ("postgres", (1,)) + assert "$1" in by_using.sql + assert (by_session.dialect, by_session.params) == ("postgres", (1,)) diff --git a/tests/test_static_contracts.py b/tests/test_static_contracts.py index fc7d1fa7..5a90135a 100644 --- a/tests/test_static_contracts.py +++ b/tests/test_static_contracts.py @@ -45,6 +45,27 @@ def test_mutating_query_methods_cannot_carry_pagination(): assert "compile_query" in called, ( f"Query.{method_name}() must build its payload via compile_query" ) + # `.sql()` runs this same body with the send swapped for its render + # twin, so the contract above covers `update.sql` / `delete.sql` too. + decorators = { + dec.id for dec in method.decorator_list if isinstance(dec, ast.Name) + } + # `update` is `_update_terminal`: `terminal` at runtime, its two forms + # overloaded for the type checker. + assert decorators & {"terminal", "_update_terminal"}, ( + f"Query.{method_name} must be a @terminal so .sql() shares its body" + ) + sends = [ + tuple(arg.id for arg in node.args if isinstance(arg, ast.Name)) + for node in ast.walk(method) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "send" + ] + assert sends == [(f"{method_name}_filtered", f"render_{method_name}")], ( + f"Query.{method_name}() must send through " + f"send({method_name}_filtered, render_{method_name})" + ) def _models_module_tree() -> ast.Module: @@ -214,6 +235,26 @@ def test_projected_shape_misuse_fails_the_type_checker(): ) +def test_each_update_form_passes_the_type_checker(): + """The recipe form and the keyword form each type-check, on `update()` + and on `update.sql()`.""" + result = _run_ty(FIXTURES / "good_updates.py") + assert result.returncode == 0, ( + f"update() forms must type-check cleanly:\n{result.stdout}\n{result.stderr}" + ) + + +def test_mixing_update_forms_fails_the_type_checker(): + """A recipe and keyword literals together is a static error on + `update()` and on `update.sql()`, not only the runtime `TypeError`.""" + result = _run_ty(FIXTURES / "bad_updates.py") + assert result.returncode != 0, "bad_updates.py must fail ty" + assert result.stdout.count("no-matching-overload") == 2, ( + "the mixed form on update() and on update.sql() must each be a " + f"static error:\n{result.stdout}" + ) + + def test_include_misuse_fails_the_type_checker(): """include() on a projected query (the `self: Never` pin, #287) and a string include selector are static errors at the call site."""