From 565b8b73de22d1b0f3b26c4d5ed70f499c137401 Mon Sep 17 00:00:00 2001 From: Taylor Date: Wed, 21 Oct 2026 02:50:14 -0400 Subject: [PATCH 1/4] refactor(operations): one build per query terminal, shared by send and render Each query terminal's statement build (plan checks, catalog read, joins, select list, order/paging, sea-query build) moves into one build_ function that the executing entry sends and a new render_ entry returns without sending. The statement log's parameter masking becomes sql_log::shown_params so a rendered statement shows parameters the way a record does. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01Hb9U9A7Zu9ypdHTsDvdScP --- src/ferro/_core.pyi | 35 ++ src/lib.rs | 4 + src/operations.rs | 1053 ++++++++++++++++++++++++++++++------------- src/sql_log.rs | 47 +- 4 files changed, 799 insertions(+), 340 deletions(-) diff --git a/src/ferro/_core.pyi b/src/ferro/_core.pyi index a5bab693..77b0ef1a 100644 --- a/src/ferro/_core.pyi +++ b/src/ferro/_core.pyi @@ -707,6 +707,41 @@ 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/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..8394e2dd 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)?; From d8af32bb295ca0e9efa0d5ae84da054912bd8eb5 Mon Sep 17 00:00:00 2001 From: Taylor Date: Wed, 21 Oct 2026 02:53:40 -0400 Subject: [PATCH 2/4] feat(query): .sql() on every Query terminal renders its statement all, first, count, exists, update (both forms) and delete become terminals: calling one runs it, and its .sql() returns the RenderedSql the call would send (sql, params, dialect) without sending it. .sql() runs the terminal's own body with its one send swapped for the render twin, so argument checks, routing and the statement build are the call's. Secrets show as REDACTED. RenderedSql and REDACTED are exported from ferro. The parity pin in tests/test_rendered_statement.py holds every terminal's rendering equal to the ferro.sql record of the executed call on both dialects, and AGENTS.md lists it under I-1's enforcement. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01Hb9U9A7Zu9ypdHTsDvdScP --- AGENTS.md | 9 + docs/pages/api/queries.md | 6 + src/ferro/__init__.py | 4 +- src/ferro/_core.pyi | 1 + src/ferro/query/__init__.py | 3 + src/ferro/query/builder.py | 122 ++++++---- src/ferro/query/rendered.py | 232 +++++++++++++++++++ tests/test_query_typing.py | 43 ++++ tests/test_rendered_statement.py | 369 +++++++++++++++++++++++++++++++ tests/test_static_contracts.py | 19 ++ 10 files changed, 761 insertions(+), 47 deletions(-) create mode 100644 src/ferro/query/rendered.py create mode 100644 tests/test_rendered_statement.py 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 77b0ef1a..e1b2251b 100644 --- a/src/ferro/_core.pyi +++ b/src/ferro/_core.pyi @@ -707,6 +707,7 @@ 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.""" 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..af36d885 100644 --- a/src/ferro/query/builder.py +++ b/src/ferro/query/builder.py @@ -7,10 +7,12 @@ """ import copy +from collections.abc import Awaitable from typing import ( TYPE_CHECKING, Any, Callable, + ClassVar, Generic, Literal, NamedTuple, @@ -30,6 +32,10 @@ delete_filtered, fetch_filtered, remove_m2m_links, + render_count, + render_delete, + render_fetch, + render_update, update_filtered, ) from .nodes import ( @@ -44,6 +50,7 @@ ValueExpr, validate_query_column, ) +from .rendered import Terminal, send, terminal from .rows import Row, Rows from .wire import ( M2mContext, @@ -1236,7 +1243,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 +1257,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 +1268,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 +1286,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 +1294,13 @@ async def _count(self, operation: Literal["count", "exists"]) -> int: operation=operation, ) - @overload + @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 +1313,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 +1342,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 +1354,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 +1376,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 +1398,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 +1731,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 +1753,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 +1817,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 +1832,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 +1844,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 +1942,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 +2007,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/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..f00a825f 100644 --- a/tests/test_static_contracts.py +++ b/tests/test_static_contracts.py @@ -45,6 +45,25 @@ 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) + } + assert "terminal" in decorators, ( + 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: From cbb0abb5a256daab239d613ed2c954faf10e5df5 Mon Sep 17 00:00:00 2001 From: Taylor Date: Wed, 21 Oct 2026 02:55:01 -0400 Subject: [PATCH 3/4] test(sql_log): shown_params masks secret binds as None Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01Hb9U9A7Zu9ypdHTsDvdScP --- src/sql_log.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/src/sql_log.rs b/src/sql_log.rs index 8394e2dd..3c8cde6d 100644 --- a/src/sql_log.rs +++ b/src/sql_log.rs @@ -385,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()); From 23e9e226db5873fbd6a2d638f6fbcc6a616baa1e Mon Sep 17 00:00:00 2001 From: Taylor Date: Wed, 21 Oct 2026 03:03:37 -0400 Subject: [PATCH 4/4] fix(query): keep update()'s two forms apart statically, on the call and .sql() Query.update is now _update_terminal: terminal() at runtime, read by a type checker through a descriptor whose bound terminal overloads __call__ and sql per form (a recipe, or keyword literals). Mixing them is a static no-matching-overload again, as it was before update became a terminal. tests/static_fixtures/bad_updates.py and good_updates.py pin both sides. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01Hb9U9A7Zu9ypdHTsDvdScP --- src/ferro/query/builder.py | 45 ++++++++++++++++++++++++++- tests/static_fixtures/bad_updates.py | 26 ++++++++++++++++ tests/static_fixtures/good_updates.py | 26 ++++++++++++++++ tests/test_static_contracts.py | 24 +++++++++++++- 4 files changed, 119 insertions(+), 2 deletions(-) create mode 100644 tests/static_fixtures/bad_updates.py create mode 100644 tests/static_fixtures/good_updates.py diff --git a/src/ferro/query/builder.py b/src/ferro/query/builder.py index af36d885..6ac11d6f 100644 --- a/src/ferro/query/builder.py +++ b/src/ferro/query/builder.py @@ -18,6 +18,7 @@ NamedTuple, Never, NoReturn, + Protocol, Self, Type, TypeVar, @@ -561,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. @@ -1294,7 +1337,7 @@ async def _count(self, operation: Literal["count", "exists"]) -> int: operation=operation, ) - @terminal + @_update_terminal async def update( self: "Query[T]", recipe: Callable[[QueryProxy[T]], dict[str, Any]] | None = 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_static_contracts.py b/tests/test_static_contracts.py index f00a825f..5a90135a 100644 --- a/tests/test_static_contracts.py +++ b/tests/test_static_contracts.py @@ -50,7 +50,9 @@ def test_mutating_query_methods_cannot_carry_pagination(): decorators = { dec.id for dec in method.decorator_list if isinstance(dec, ast.Name) } - assert "terminal" in decorators, ( + # `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 = [ @@ -233,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."""