umbral_core/orm/queryset/mod.rs
1//! `QuerySet<T>` and `Manager<T>`: chainable lazy SQL builder + entry
2//! point.
3//!
4//! `T::objects()` returns a `Manager<T>`; chaining `.filter` / `.order_by`
5//! / `.limit` / etc. on it (or on a `QuerySet<T>` directly) yields a new
6//! `QuerySet<T>`. Terminals (`.fetch`, `.first`, `.count`, `.exists`)
7//! await an async DB roundtrip via the ambient or explicit pool.
8//!
9//! At M1 the surface is intentionally narrow per
10//! `docs/specs/03-orm-querysets.md`: filter / order_by / limit / offset
11//! for chaining, and fetch / first / count / exists for terminals. No
12//! exclude / distinct / values / annotate / aggregate / update / delete
13//! yet — those land as later milestones surface real need.
14//!
15//! M2 lifted the terminals and the `Manager` delegation onto a generic
16//! `T: Model` bound. The table name comes from `T::TABLE`, the SELECT
17//! column list from `T::FIELDS`, and row materialisation from the
18//! `FromRow` bound the terminals carry. M3 generates the `Model` impl
19//! from `#[derive(Model)]`.
20//!
21//! ## Phase 2.5 — backend-agnostic terminals
22//!
23//! Through Phase 2 the QuerySet stored a `SqlitePool` and built every
24//! query with sea-query's `SqliteQueryBuilder`. Phase 2.5 widens that:
25//! the explicit-pool slot is `Option<DbPool>`, `.on(&SqlitePool)` keeps
26//! working unchanged, and a new `.on_pg(&PgPool)` registers a Postgres
27//! pool. The terminal methods dispatch on the resolved pool variant —
28//! SQLite path uses `SqliteQueryBuilder` + a `SqlitePool` executor;
29//! Postgres path uses `PostgresQueryBuilder` + a `PgPool` executor.
30//!
31//! The row-materialization bound on each terminal is the conjunction
32//! of both backends' `FromRow` impls. `#[derive(sqlx::FromRow)]` emits
33//! a generic-over-`R` impl, so a user struct with standard field
34//! types satisfies both bounds without any per-backend ceremony.
35
36mod backend_pg;
37mod backend_sqlite;
38mod errors;
39pub(crate) mod hydration;
40mod tx;
41mod write_helpers;
42
43pub use errors::{GetError, TryForEachError};
44use hydration::{hydrate_prefetch_related, hydrate_select_related};
45pub use tx::QuerySetTx;
46use write_helpers::{
47 build_insert_many_for, build_insert_one_for, fk_pk_hint, pk_field, serialize_to_map,
48};
49
50use std::collections::HashMap;
51use std::marker::PhantomData;
52
53use sea_query::{
54 Alias, Expr, Func, IntoIden, Order, PostgresQueryBuilder, Query, SqliteQueryBuilder,
55};
56use sea_query_binder::SqlxBinder;
57use serde_json::Value as JsonValue;
58
59use crate::db::DbPool;
60use crate::orm::{FExpr, HydrateRelated, Model, OrderExpr, Predicate};
61use umbral_casing::to_snake_case;
62
63/// Entry point for queries on a model.
64///
65/// `Manager<T>` wraps a freshly-constructed `QuerySet<T>` and exposes
66/// the same chainable surface. The user never constructs one directly;
67/// `Post::objects()` is the only door.
68pub struct Manager<T> {
69 _phantom: PhantomData<T>,
70 /// Per-Manager override for the `atomic_transactions` builder
71 /// default. `None` = inherit the global default; `Some(true)` =
72 /// wrap subsequent writes in a transaction; `Some(false)` =
73 /// explicitly opt out. Propagates into the QuerySet `queryset()`
74 /// constructs.
75 atomic: Option<bool>,
76}
77
78impl<T> Manager<T> {
79 pub(crate) fn new() -> Self {
80 Self {
81 _phantom: PhantomData,
82 atomic: None,
83 }
84 }
85
86 /// Wrap every write terminal that hangs off this Manager in a
87 /// transaction. Equivalent to calling `.atomic()` on each
88 /// QuerySet derived from it. Per-call `.non_atomic()` overrides.
89 pub fn atomic(mut self) -> Self {
90 self.atomic = Some(true);
91 self
92 }
93
94 /// Opt this Manager (and every QuerySet derived from it) out of
95 /// the global `App::builder().atomic_transactions(true)` default.
96 pub fn non_atomic(mut self) -> Self {
97 self.atomic = Some(false);
98 self
99 }
100}
101
102impl<T> Default for Manager<T> {
103 fn default() -> Self {
104 Self::new()
105 }
106}
107
108/// SQL join flavor recorded per `join_related` hop. `None` in a
109/// `JoinReq` means "infer from FK nullability" (gap 4c); an explicit
110/// `left_/inner_/right_join_related` records `Some(..)`.
111#[derive(Debug, Clone, Copy, PartialEq, Eq)]
112pub enum JoinKind {
113 Inner,
114 Left,
115 Right,
116}
117
118impl JoinKind {
119 /// Lower to sea-query's join type.
120 pub(crate) fn sea(self) -> sea_query::JoinType {
121 match self {
122 JoinKind::Inner => sea_query::JoinType::InnerJoin,
123 JoinKind::Left => sea_query::JoinType::LeftJoin,
124 JoinKind::Right => sea_query::JoinType::RightJoin,
125 }
126 }
127}
128
129/// One requested eager-join: a dotted relation path (`"plugin__author"`)
130/// plus the join type to apply to the LAST hop. `kind: None` means
131/// auto-infer per-hop from FK nullability (INNER for NOT NULL, LEFT for
132/// nullable) — Django's default. The explicit methods pin `Some(..)`.
133#[derive(Debug, Clone)]
134pub(crate) struct JoinReq {
135 pub(crate) path: String,
136 pub(crate) kind: Option<JoinKind>,
137}
138
139/// A lazy, chainable SQL query.
140///
141/// Carries a sea-query `SelectStatement` plus pool-resolution state.
142/// Nothing is sent to the database until a terminal method is awaited.
143/// Cloning is cheap (the `SelectStatement` clones in O(query size)).
144pub struct QuerySet<T> {
145 /// The base SelectStatement — FROM, columns, joins, group-by,
146 /// order-by, limit, offset. Filters are NOT applied here; they
147 /// accumulate on [`Self::predicates`] and get woven in at
148 /// terminal time, so per-backend predicate variants (Phase
149 /// 4.2.2) can pick the right SimpleExpr based on the resolved
150 /// pool.
151 pub(crate) query: sea_query::SelectStatement,
152 /// Accumulated filter predicates. Each one renders to either its
153 /// default `cond` (for Postgres) or its `cond_sqlite` override
154 /// (for SQLite, if set) at terminal time.
155 pub(crate) predicates: Vec<Predicate<T>>,
156 pub(crate) explicit_pool: Option<DbPool>,
157 /// FK field names requested for eager loading via `select_related`.
158 /// After the main query returns rows, a batch `IN (...)` query
159 /// fetches the related rows for each named field and calls
160 /// `HydrateRelated::hydrate_fk` to populate `ForeignKey.resolved`.
161 pub(crate) select_related: Vec<String>,
162 /// M2M field names requested for eager loading via
163 /// `prefetch_related`. After the main query, one batched JOIN
164 /// against the junction + child table fetches every related row
165 /// for every parent in a single round-trip; each parent's
166 /// `M2M.resolved` slot is populated via
167 /// `HydrateRelated::set_m2m_resolved_json`. Gap #19.
168 pub(crate) prefetch_related: Vec<String>,
169 /// BUG-8: `#[umbral(ordering = [...])]` lowers to a default ORDER
170 /// BY applied at terminal time when the caller didn't supply an
171 /// explicit `.order_by(...)`. Mirrors Django's
172 /// `Meta.ordering` semantics: explicit calls REPLACE the default
173 /// rather than appending to it.
174 pub(crate) default_ordering: Vec<(&'static str, bool)>,
175 /// Set to `true` the first time `.order_by(...)` is called; when
176 /// `false`, `build_query_for` applies `default_ordering`.
177 pub(crate) explicit_order: bool,
178 /// Per-QuerySet override for the `atomic_transactions` builder
179 /// default. `None` = inherit the global default via
180 /// [`crate::db::atomic_default`]; `Some(true)` = wrap this
181 /// QuerySet's write terminal in a transaction; `Some(false)` =
182 /// explicitly opt out.
183 pub(crate) atomic: Option<bool>,
184 /// Feature #72 — soft-delete state. Snapshotted from
185 /// `T::SOFT_DELETE` when the QuerySet is constructed from a
186 /// Manager (the no-bounds `QuerySet::new` constructor leaves
187 /// this `false` so hand-built QuerySets stay opt-out by
188 /// default).
189 pub(crate) soft_delete_active: bool,
190 /// True when the caller opted back into soft-deleted rows via
191 /// `.with_deleted()`. Skips the auto `WHERE deleted_at IS NULL`
192 /// injection.
193 pub(crate) with_deleted: bool,
194 /// True when the caller wants ONLY soft-deleted rows via
195 /// `.only_deleted()`. Inverts the auto-filter to
196 /// `WHERE deleted_at IS NOT NULL`.
197 pub(crate) only_deleted: bool,
198 /// True when the caller asked for a real DELETE via
199 /// `.hard_delete()` — bypasses the soft-delete rewrite that
200 /// would normally turn `delete()` into an UPDATE.
201 pub(crate) hard_delete: bool,
202 /// Gap #111 — column projection set by [`Self::only`]. When
203 /// `Some`, [`Self::to_sql`] / [`Self::to_sql_pg`] swap the
204 /// SELECT list for just these columns, and the typed terminals
205 /// (`fetch` / `first` / `get`) refuse to run with a clear error
206 /// pointing the caller at [`Self::values`] (FromRow can't
207 /// hydrate `T` from a partial-column row). `None` keeps the
208 /// pre-#111 behaviour (full SELECT).
209 pub(crate) only_cols: Option<Vec<String>>,
210 /// FK field names requested for JOIN-based prefetch via
211 /// [`Self::join_related`]. Distinct from `select_related`:
212 /// `join_related` weaves `LEFT JOIN <related_table>` into the
213 /// main SELECT (with aliased columns `<field>__<col>`) so one
214 /// round-trip pulls parent + related rows together. The existing
215 /// `select_related` path keeps its "batched-IN-followup-query"
216 /// shape — both are valid; this one wins when round-trip count
217 /// matters more than the per-row column overhead.
218 pub(crate) join_related: Vec<JoinReq>,
219 /// Related-aggregate annotations added via
220 /// [`Self::annotate_related`] / [`Self::annotate_count`]. Applied
221 /// inside `build_query_for`, so EVERY terminal and introspection
222 /// path — `fetch_annotated`, `explain`, `to_sql`, `to_sql_pg` —
223 /// sees the same correlated subqueries. That's the Django
224 /// `annotate()` contract: an annotation is query-builder state,
225 /// not a side query.
226 pub(crate) annotations: Vec<RelatedAnnotation>,
227 _phantom: PhantomData<T>,
228}
229
230// Manual `Clone` — NOT `#[derive(Clone)]`, because the derive would force a
231// spurious `T: Clone` bound (every field is `T`-independent or carries its
232// own `T`-free `Clone`: `Predicate<T>` has a manual `impl<T> Clone`, and
233// `PhantomData<T>` clones for any `T`). The doc comment above already
234// promised cheap cloning; this is what makes `Paginator` (and any
235// requery-without-consume caller) slice the same query per page.
236impl<T> Clone for QuerySet<T> {
237 fn clone(&self) -> Self {
238 Self {
239 query: self.query.clone(),
240 predicates: self.predicates.clone(),
241 explicit_pool: self.explicit_pool.clone(),
242 select_related: self.select_related.clone(),
243 prefetch_related: self.prefetch_related.clone(),
244 default_ordering: self.default_ordering.clone(),
245 explicit_order: self.explicit_order,
246 atomic: self.atomic,
247 soft_delete_active: self.soft_delete_active,
248 with_deleted: self.with_deleted,
249 only_deleted: self.only_deleted,
250 hard_delete: self.hard_delete,
251 only_cols: self.only_cols.clone(),
252 join_related: self.join_related.clone(),
253 annotations: self.annotations.clone(),
254 _phantom: PhantomData,
255 }
256 }
257}
258
259// Manual `Debug` for the same `T: Debug`-free reason; `Predicate<T>`'s own
260// `Debug` is likewise `T`-free.
261impl<T> std::fmt::Debug for QuerySet<T> {
262 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
263 // `Predicate<T>` is intentionally not `Debug` (its `SimpleExpr`
264 // carries no `T`), so report the predicate count rather than the
265 // opaque expressions.
266 f.debug_struct("QuerySet")
267 .field("query", &self.query)
268 .field("predicates", &format_args!("[{} predicate(s)]", self.predicates.len()))
269 .field("explicit_pool", &self.explicit_pool)
270 .field("select_related", &self.select_related)
271 .field("prefetch_related", &self.prefetch_related)
272 .field("default_ordering", &self.default_ordering)
273 .field("explicit_order", &self.explicit_order)
274 .field("atomic", &self.atomic)
275 .field("soft_delete_active", &self.soft_delete_active)
276 .field("with_deleted", &self.with_deleted)
277 .field("only_deleted", &self.only_deleted)
278 .field("hard_delete", &self.hard_delete)
279 .field("only_cols", &self.only_cols)
280 .field("join_related", &self.join_related)
281 .field("annotations", &self.annotations)
282 .finish()
283 }
284}
285
286/// One related-aggregate annotation on a [`QuerySet`] —
287/// `(alias, relation, aggregate)` resolved against the model's
288/// `REVERSE_FK_RELATIONS` at builder time. A name that fails to
289/// resolve is stored poisoned (`resolved: Err`) so the infallible
290/// builder stays chainable while every fallible consumer
291/// (`fetch_annotated`, `explain`) reports it loudly.
292#[derive(Debug, Clone)]
293pub(crate) struct RelatedAnnotation {
294 pub(crate) alias: String,
295 pub(crate) agg: crate::orm::Aggregate,
296 /// `Ok((child_table, fk_column, parent_table, parent_pk))` or the
297 /// loud error message for an unknown relation.
298 pub(crate) resolved: Result<(String, String, String, String), String>,
299 /// Child model is `#[umbral(soft_delete)]` — fold
300 /// `AND <child>.deleted_at IS NULL` into the correlated subquery so
301 /// a trashed child stops inflating the parent's count.
302 pub(crate) child_soft_delete: bool,
303 /// Optional child-side predicate (Django's `Count(filter=Q(...))`),
304 /// pre-rendered to a backend-default `SimpleExpr`. ANDed into the
305 /// subquery WHERE. From `annotate_count_where`.
306 pub(crate) child_filter: Option<sea_query::SimpleExpr>,
307 /// `Some(junction_table)` when this annotation counts M2M junction
308 /// rows instead of child rows (`annotate_count` over an `M2M<T>`).
309 pub(crate) m2m_junction: Option<String>,
310}
311
312/// Outcome of auto-discovering a reverse-FK relation (gaps2 #45) when
313/// the parent declares no matching `ReverseSet` field. The resolver
314/// scans the registry for children whose FK targets the parent table
315/// and matches `relation` against their conventional name forms.
316enum AutoDiscovery {
317 /// Exactly one (child, fk_column) candidate matched.
318 Resolved {
319 child_table: String,
320 fk_column: String,
321 soft_delete: bool,
322 },
323 /// Two or more candidates matched — the caller must declare a
324 /// `#[umbral(reverse_fk = "...")]` field to disambiguate. Carries
325 /// the candidate `child.fk` labels for the error message.
326 Ambiguous(Vec<String>),
327 /// No candidate matched. Carries the list of auto-discoverable
328 /// child names so the error can teach the available relations.
329 NotFound(Vec<String>),
330}
331
332// `snake_case` replaced by `umbral_casing::to_snake_case` (imported above)
333// in the gaps2 #77 consolidation refactor.
334
335/// Scan the model registry for children whose FK targets `T::TABLE`,
336/// and match `relation` against each candidate's conventional name
337/// forms: the child's table name, the child's struct name in
338/// snake_case and bare-lowercase, and any of those with a `_set`
339/// suffix (Django's `<model>_set`). Declared `REVERSE_FK_RELATIONS` /
340/// `M2M_RELATIONS` are resolved by the caller BEFORE this runs, so
341/// they always take precedence.
342fn discover_reverse_relation<T: crate::orm::Model>(relation: &str) -> AutoDiscovery {
343 if !crate::migrate::is_initialised() {
344 return AutoDiscovery::NotFound(Vec::new());
345 }
346 let parent_table = T::TABLE;
347 // Each candidate: (child_table, fk_column, child_soft_delete).
348 let mut candidates: Vec<(String, String, bool)> = Vec::new();
349 let mut discoverable: Vec<String> = Vec::new();
350 for meta in crate::migrate::registered_models() {
351 for col in &meta.fields {
352 if col.fk_target.as_deref() != Some(parent_table) {
353 continue;
354 }
355 // Conventional name forms this (child, fk_column) answers to.
356 let snake = to_snake_case(&meta.name);
357 let lower = meta.name.to_ascii_lowercase();
358 let mut forms = vec![
359 meta.table.clone(),
360 snake.clone(),
361 lower.clone(),
362 format!("{}_set", meta.table),
363 format!("{snake}_set"),
364 format!("{lower}_set"),
365 ];
366 forms.sort();
367 forms.dedup();
368 // Surface a friendly name for "available children" errors.
369 discoverable.push(format!("{}_set", meta.table));
370 if forms.iter().any(|f| f == relation) {
371 candidates.push((meta.table.clone(), col.name.clone(), meta.soft_delete));
372 }
373 }
374 }
375 match candidates.len() {
376 1 => {
377 let (child_table, fk_column, soft_delete) = candidates.pop().unwrap();
378 AutoDiscovery::Resolved {
379 child_table,
380 fk_column,
381 soft_delete,
382 }
383 }
384 0 => {
385 discoverable.sort();
386 discoverable.dedup();
387 AutoDiscovery::NotFound(discoverable)
388 }
389 _ => {
390 let labels = candidates
391 .into_iter()
392 .map(|(child, fk, _)| format!("{child}.{fk}"))
393 .collect();
394 AutoDiscovery::Ambiguous(labels)
395 }
396 }
397}
398
399impl<T> QuerySet<T> {
400 pub(crate) fn new(query: sea_query::SelectStatement) -> Self {
401 Self {
402 query,
403 default_ordering: Vec::new(),
404 explicit_order: false,
405 predicates: Vec::new(),
406 explicit_pool: None,
407 select_related: Vec::new(),
408 prefetch_related: Vec::new(),
409 atomic: None,
410 soft_delete_active: false,
411 with_deleted: false,
412 only_deleted: false,
413 hard_delete: false,
414 only_cols: None,
415 join_related: Vec::new(),
416 annotations: Vec::new(),
417 _phantom: PhantomData,
418 }
419 }
420
421 /// Feature #72 — include soft-deleted rows in this query. Skips
422 /// the auto `WHERE deleted_at IS NULL` injection. No-op on
423 /// models that aren't tagged `#[umbral(soft_delete)]`.
424 pub fn with_deleted(mut self) -> Self {
425 self.with_deleted = true;
426 self
427 }
428
429 /// Feature #72 — only soft-deleted rows. Useful for admin
430 /// trash views and undelete workflows. No-op on models that
431 /// aren't tagged `#[umbral(soft_delete)]`.
432 pub fn only_deleted(mut self) -> Self {
433 self.only_deleted = true;
434 self
435 }
436
437 /// Feature #72 — force a real DELETE for the next `.delete()`
438 /// terminal call. Soft-delete models normally rewrite delete()
439 /// as `UPDATE ... SET deleted_at = NOW()`; `.hard_delete()`
440 /// bypasses that for GDPR purges, test cleanup, or any other
441 /// case where the row truly should be gone. No-op on models
442 /// that aren't tagged `#[umbral(soft_delete)]` (their delete()
443 /// is already a hard DELETE).
444 pub fn hard_delete(mut self) -> Self {
445 self.hard_delete = true;
446 self
447 }
448
449 /// Gap #111 — restrict the SELECT to the named columns.
450 ///
451 /// Affects [`Self::to_sql`] / [`Self::to_sql_pg`] (the SELECT list
452 /// shrinks to just these columns) and propagates into
453 /// [`Self::values`] when that terminal is called without its own
454 /// explicit column slice.
455 ///
456 /// **The typed terminals (`fetch` / `first` / `get`) refuse to
457 /// run with `.only()` set** because a partial-column row can't
458 /// satisfy `T`'s `FromRow` impl. The error message points at
459 /// `.values(...)` (returns `Vec<serde_json::Value>`) as the
460 /// execution path. Use `.only()` for `.to_sql()` inspection and
461 /// `.values()` for actual reads.
462 ///
463 /// ```rust,ignore
464 /// // Inspect: "SELECT \"id\", \"name\" FROM \"brand\" WHERE \"id\" = ?"
465 /// let sql = Brand::objects()
466 /// .filter(brand::ID.eq(1))
467 /// .only(&["id", "name"])
468 /// .to_sql();
469 ///
470 /// // Execute (returns JSON rows):
471 /// let rows = Brand::objects()
472 /// .filter(brand::ID.eq(1))
473 /// .values(&["id", "name"])
474 /// .await?;
475 /// ```
476 ///
477 /// Unknown column names are not validated here — they surface at
478 /// terminal time the same way `.values()` reports them (the
479 /// rendered SQL contains the bad identifier and SQLite/Postgres
480 /// raises). This keeps the chainable surface return-type-stable.
481 pub fn only(mut self, cols: &[&str]) -> Self {
482 self.only_cols = Some(cols.iter().map(|s| s.to_string()).collect());
483 self
484 }
485
486 /// Wrap this QuerySet's write terminal in a transaction. Reads are
487 /// unaffected (read terminals are single statements and the DB
488 /// gives them a consistent snapshot). Mutually exclusive with
489 /// [`Self::on_tx`] — if both are set, `on_tx` wins (you're
490 /// already inside a transaction, so wrapping again would deadlock
491 /// or fail on backends without nested transactions).
492 pub fn atomic(mut self) -> Self {
493 self.atomic = Some(true);
494 self
495 }
496
497 /// Opt this QuerySet's write terminal out of the global
498 /// `App::builder().atomic_transactions(true)` default. Useful in
499 /// hot-path batches where the caller already owns the outer
500 /// transaction.
501 pub fn non_atomic(mut self) -> Self {
502 self.atomic = Some(false);
503 self
504 }
505
506 /// Resolve whether this QuerySet should auto-wrap its write
507 /// terminal in a transaction. Per-call override > builder global.
508 pub(crate) fn should_atomic_wrap(&self) -> bool {
509 self.atomic.unwrap_or_else(crate::db::atomic_default)
510 }
511
512 /// Clone the base query and weave in the accumulated predicates,
513 /// picking the dialect-appropriate `SimpleExpr` for each one. The
514 /// `backend_name` is `"sqlite"` or `"postgres"`; any other value
515 /// behaves like Postgres (the default).
516 pub(crate) fn build_query_for(&self, backend_name: &str) -> sea_query::SelectStatement {
517 let mut q = self.query.clone();
518 for p in &self.predicates {
519 q.and_where(p.cond_for(backend_name));
520 }
521 // Feature #72 — soft-delete auto-filter. When the model
522 // opted in via `#[umbral(soft_delete)]` AND the caller
523 // didn't switch the visibility via `.with_deleted()` /
524 // `.only_deleted()`, inject `WHERE deleted_at IS NULL`.
525 // `.with_deleted()` shows everything; `.only_deleted()`
526 // shows just the soft-deleted rows.
527 if self.soft_delete_active {
528 use sea_query::Expr;
529 if self.only_deleted {
530 q.and_where(Expr::col(Alias::new("deleted_at")).is_not_null());
531 } else if !self.with_deleted {
532 q.and_where(Expr::col(Alias::new("deleted_at")).is_null());
533 }
534 }
535 // Related-aggregate annotations: one correlated scalar
536 // subquery per entry, aliased onto the SELECT list. Living
537 // HERE is what makes `.annotate_*` compose with everything —
538 // explain(), to_sql(), fetch_annotated() all see the same
539 // query. Poisoned entries (unknown relation) are skipped in
540 // this infallible path; the fallible consumers call
541 // `check_annotations()` first and fail loudly instead.
542 for ann in &self.annotations {
543 // M2M-junction annotations count rows of the junction table
544 // (`<parent>_<field>`, columns parent_id / child_id),
545 // correlated on parent_id = <parent>.<pk>.
546 if let Some(junction) = &ann.m2m_junction {
547 if let Ok((_child_table, _fk_col, parent_table, parent_pk)) = &ann.resolved {
548 let mut sub = sea_query::Query::select();
549 sub.expr(ann.agg.to_simple_expr())
550 .from(crate::db::router::schema_qualified_table(junction.as_str()))
551 .and_where(
552 sea_query::Expr::col((
553 Alias::new(junction.as_str()),
554 Alias::new("parent_id"),
555 ))
556 .equals((
557 Alias::new(parent_table.as_str()),
558 Alias::new(parent_pk.as_str()),
559 )),
560 );
561 q.expr_as(
562 sea_query::SimpleExpr::SubQuery(
563 None,
564 Box::new(sea_query::SubQueryStatement::SelectStatement(
565 sub.to_owned(),
566 )),
567 ),
568 Alias::new(ann.alias.as_str()),
569 );
570 }
571 continue;
572 }
573 if let Ok((child_table, fk_col, parent_table, parent_pk)) = &ann.resolved {
574 let mut sub = sea_query::Query::select();
575 sub.expr(ann.agg.to_simple_expr())
576 .from(crate::db::router::schema_qualified_table(
577 child_table.as_str(),
578 ))
579 .and_where(
580 sea_query::Expr::col((
581 Alias::new(child_table.as_str()),
582 Alias::new(fk_col.as_str()),
583 ))
584 .equals((
585 Alias::new(parent_table.as_str()),
586 Alias::new(parent_pk.as_str()),
587 )),
588 );
589 // Auto-exclude soft-deleted children from the count when
590 // the child model is `#[umbral(soft_delete)]`.
591 if ann.child_soft_delete {
592 sub.and_where(
593 sea_query::Expr::col((
594 Alias::new(child_table.as_str()),
595 Alias::new("deleted_at"),
596 ))
597 .is_null(),
598 );
599 }
600 // Child-side predicate (annotate_count_where).
601 if let Some(filter) = &ann.child_filter {
602 sub.and_where(filter.clone());
603 }
604 q.expr_as(
605 sea_query::SimpleExpr::SubQuery(
606 None,
607 Box::new(sea_query::SubQueryStatement::SelectStatement(
608 sub.to_owned(),
609 )),
610 ),
611 Alias::new(ann.alias.as_str()),
612 );
613 }
614 }
615 // BUG-8: default ORDER BY applies only when the caller didn't
616 // supply an explicit `.order_by(...)`. Mirrors Django's
617 // `Meta.ordering` semantics.
618 if !self.explicit_order {
619 for (col, desc) in &self.default_ordering {
620 let order = if *desc { Order::Desc } else { Order::Asc };
621 q.order_by(Alias::new(*col), order);
622 }
623 }
624 q
625 }
626}
627
628/// Chainable methods on every `QuerySet<T>`.
629///
630/// These are model-agnostic: they only touch the sea-query
631/// `SelectStatement` and the pool-resolution slot, neither of which
632/// depends on `T`. Terminals (which need row mapping) live in the
633/// `impl<T: Model> QuerySet<T>` block below.
634impl<T> QuerySet<T> {
635 /// Add a WHERE condition. Multiple `.filter` calls AND together
636 /// (sea-query's `and_where` semantics — applied at terminal time
637 /// once the resolved pool's backend is known).
638 pub fn filter(mut self, p: Predicate<T>) -> Self {
639 self.predicates.push(p);
640 self
641 }
642
643 /// Add a negated WHERE condition. The negated predicate ANDs into
644 /// the chain alongside any `filter()` calls, so
645 /// `.filter(A).exclude(B).filter(C)` renders as `WHERE A AND NOT B
646 /// AND C`. Sugar for `filter(Q::not(p))`.
647 ///
648 /// Mirrors Django's `QuerySet.exclude()`.
649 pub fn exclude(self, p: Predicate<T>) -> Self {
650 self.filter(crate::orm::Q::not(p))
651 }
652
653 /// Add an ORDER BY clause. Multiple `.order_by` calls append.
654 /// The first explicit call also opts out of the model's
655 /// `#[umbral(ordering = [...])]` default (BUG-8) — Django semantics:
656 /// explicit ordering replaces the default rather than stacking on
657 /// top of it.
658 pub fn order_by(mut self, o: OrderExpr<T>) -> Self {
659 let order = if o.descending {
660 Order::Desc
661 } else {
662 Order::Asc
663 };
664 self.query.order_by(Alias::new(o.column), order);
665 self.explicit_order = true;
666 self
667 }
668
669 /// Set LIMIT.
670 pub fn limit(mut self, n: u64) -> Self {
671 self.query.limit(n);
672 self
673 }
674
675 /// Set OFFSET.
676 pub fn offset(mut self, n: u64) -> Self {
677 self.query.offset(n);
678 self
679 }
680
681 /// Override the pool resolved at terminal time with a SQLite pool.
682 ///
683 /// Wins over the ambient default. Used by tests that drive the ORM
684 /// without going through `App::build()`. For a Postgres override
685 /// use [`Self::on_pg`].
686 pub fn on(mut self, pool: &sqlx::SqlitePool) -> Self {
687 self.explicit_pool = Some(DbPool::Sqlite(pool.clone()));
688 self
689 }
690
691 /// Override the pool resolved at terminal time with a Postgres pool.
692 ///
693 /// The Postgres counterpart of [`Self::on`]. Tests that want to
694 /// exercise the Postgres branch (or that drive against a real
695 /// Postgres instance) reach for this directly.
696 pub fn on_pg(mut self, pool: &sqlx::PgPool) -> Self {
697 self.explicit_pool = Some(DbPool::Postgres(pool.clone()));
698 self
699 }
700
701 /// Attach this `QuerySet` to an open transaction.
702 ///
703 /// Returns a [`QuerySetTx`] that holds both the query and a mutable
704 /// reference to the transaction. Every terminal on `QuerySetTx`
705 /// (`fetch`, `first`, `count`, `exists`, `get`, `delete`,
706 /// `update_values`) executes inside the open transaction so all
707 /// operations in the same closure commit or roll back as a unit.
708 ///
709 /// ```rust,ignore
710 /// umbral::db::transaction(|tx| async move {
711 /// let order = Order::objects().on_tx(tx).create(new_order).await?;
712 /// Stock::objects()
713 /// .on_tx(tx)
714 /// .filter(stock::SKU.eq(sku))
715 /// .update_values(delta)
716 /// .await?;
717 /// Ok::<_, MyError>(order)
718 /// }).await?;
719 /// ```
720 pub fn on_tx(self, tx: &mut crate::db::Transaction) -> QuerySetTx<'_, T> {
721 QuerySetTx { qs: self, tx }
722 }
723
724 /// Eagerly load a single FK field by name.
725 ///
726 /// After the main SELECT returns rows, a batch `SELECT ... FROM <related_table>
727 /// WHERE id IN (...)` fetches all referenced rows in one round-trip. Each
728 /// returned row is deserialised as the target model and stored in
729 /// `ForeignKey<U>.resolved` so template rendering (`{{ post.author.username }}`)
730 /// and `serde_json::to_value(&post)["author"]["username"]` both work without
731 /// additional queries.
732 ///
733 /// Calling `select_related` multiple times accumulates the names:
734 /// `.select_related("author").select_related("editor")` works the same as
735 /// `.select_related_many(&["author", "editor"])`.
736 ///
737 /// ## Nested traversal (post-#42)
738 ///
739 /// `.select_related("author__manager")` walks the FK chain through
740 /// the `__` separator. One batched `IN (...)` query per hop —
741 /// `1 + len(hops)` round-trips regardless of parent count. No
742 /// N+1. Each hop's related row is embedded into the prior level's
743 /// JSON, and recursive `ForeignKey<T>::Deserialize` unpacks the
744 /// chain into `resolved()` slots at every depth. Bonus: a
745 /// select_related'd model now round-trips through
746 /// `serde_json::to_value(&t)` / `from_value` without losing the
747 /// resolved relation.
748 ///
749 /// ## Companion shapes
750 ///
751 /// - **`join_related(name)`** — same goal (load related rows) via
752 /// a true `LEFT JOIN` in the main SELECT. One round-trip total
753 /// vs. `select_related`'s `1 + N` batched-IN approach. Wider
754 /// per-row payload; better when round-trip count dominates.
755 /// - **`prefetch_related(name)`** — M2M batched loading (one
756 /// query per declared M2M field). For reverse-FK collections
757 /// (`prefetch_related("comment_set")`-style) see gap #44 — not
758 /// yet implemented.
759 ///
760 /// ## Loud errors
761 ///
762 /// Unknown field names (typos, M2M names accidentally passed
763 /// here, fields without `fk_target`) return a clear
764 /// `sqlx::Error::Protocol` from the terminal naming the bad hop
765 /// and the table it was looked up against. Pre-#42 these
766 /// silently no-op'd.
767 pub fn select_related(mut self, field_name: impl Into<String>) -> Self {
768 self.select_related.push(field_name.into());
769 self
770 }
771
772 /// Eagerly load multiple FK fields in one call.
773 ///
774 /// Sugar for chained `.select_related(name)` calls.
775 pub fn select_related_many(mut self, field_names: &[&str]) -> Self {
776 for name in field_names {
777 self.select_related.push(name.to_string());
778 }
779 self
780 }
781
782 /// JOIN-based eager FK loading — emits `LEFT JOIN <related> ON ...`
783 /// in the main SELECT (with aliased child columns `<field>__<col>`)
784 /// so one round-trip pulls the parent + related rows together.
785 ///
786 /// Trade-off vs. [`Self::select_related`]:
787 /// - `select_related` runs ONE extra batched query after the
788 /// main fetch (`SELECT * FROM related WHERE id IN (...)`).
789 /// Two round-trips total; rows stay narrow.
790 /// - `join_related` runs the main query AS the join — one
791 /// round-trip total — at the cost of a wider per-row payload
792 /// (every related column rides along even for duplicated
793 /// parents).
794 ///
795 /// Both populate `ForeignKey<U>.resolved` the same way so
796 /// downstream code (templates, serde) doesn't care which path
797 /// was used. Pick `join_related` when round-trip count dominates
798 /// (hot listing pages, small related tables) and `select_related`
799 /// when the related row is wide or only loaded for a subset of
800 /// the parent rows.
801 ///
802 /// Composes with `.select_related(other_fk)` and
803 /// `.prefetch_related(m2m)` — different fields can take different
804 /// paths in the same query.
805 ///
806 /// Multi-hop FK chains are supported: a `"__"`-separated path like
807 /// `"author__manager"` resolves one JOIN per hop in a single query.
808 ///
809 /// **Constraints**: FK fields must live in `model.fields` (M2M links
810 /// route through `prefetch_related`), and every related model along the
811 /// chain must be registered with the framework
812 /// (`App::builder().model::<U>()` or contributed by a plugin) so we can
813 /// resolve its column layout for the aliased SELECT.
814 pub fn join_related(mut self, field_name: impl Into<String>) -> Self {
815 self.join_related.push(JoinReq {
816 path: field_name.into(),
817 kind: None,
818 });
819 self
820 }
821
822 /// Sugar for chained [`Self::join_related`] calls.
823 pub fn join_related_many(mut self, field_names: &[&str]) -> Self {
824 for name in field_names {
825 self.join_related.push(JoinReq {
826 path: (*name).to_string(),
827 kind: None,
828 });
829 }
830 self
831 }
832
833 /// `LEFT JOIN` the related path — keeps parent rows whose relation
834 /// is absent (the relation hydrates as unresolved/None). Accepts a
835 /// nested path (`"plugin__author"`); the join type applies to the
836 /// deepest hop.
837 pub fn left_join_related(mut self, path: impl Into<String>) -> Self {
838 self.join_related.push(JoinReq {
839 path: path.into(),
840 kind: Some(JoinKind::Left),
841 });
842 self
843 }
844
845 /// `INNER JOIN` the related path — drops parent rows whose relation
846 /// is absent. Django's default for a NOT NULL FK.
847 pub fn inner_join_related(mut self, path: impl Into<String>) -> Self {
848 self.join_related.push(JoinReq {
849 path: path.into(),
850 kind: Some(JoinKind::Inner),
851 });
852 self
853 }
854
855 /// `RIGHT JOIN` the related path. Postgres-unconditional; SQLite
856 /// needs >= 3.39 — a runtime warning fires on older SQLite (see the
857 /// boot/runtime note in the joins docs). The precise version gate
858 /// lives at execute time (the SQLite driver's own error); the warn
859 /// is the early nudge.
860 pub fn right_join_related(mut self, path: impl Into<String>) -> Self {
861 self.join_related.push(JoinReq {
862 path: path.into(),
863 kind: Some(JoinKind::Right),
864 });
865 self
866 }
867
868 /// Eagerly load an M2M relation via a single batched join.
869 ///
870 /// After the main SELECT returns rows, one query of the shape
871 /// `SELECT j.parent_id, child.* FROM <child_table> child INNER
872 /// JOIN <junction> j ON child.<pk> = j.child_id WHERE
873 /// j.parent_id IN (...)` fetches every related child for every
874 /// parent in one round-trip. Each parent's `M2M.resolved` slot
875 /// is populated with its matching children.
876 ///
877 /// The M2M counterpart of [`Self::select_related`] for FKs —
878 /// same goal of killing N+1. Mirrors Django's
879 /// `prefetch_related('tags')`.
880 ///
881 /// ## Reverse-FK collections (post-#44)
882 ///
883 /// `prefetch_related` also loads `ReverseSet<C>` fields — the
884 /// "for each Post, give me every Comment that points at it"
885 /// shape. Declare the field on the parent with
886 /// `#[sqlx(skip)] #[serde(skip)]
887 /// #[umbral(reverse_fk = "<fk_col>")] pub <name>: ReverseSet<C>`
888 /// where `<fk_col>` names the FK column on `C` pointing back.
889 /// One `SELECT * FROM <child> WHERE <fk_col> IN (parent_pks)`
890 /// regardless of parent count — no N+1.
891 ///
892 /// ## Scope (v1)
893 ///
894 /// - **M2M + reverse-FK only.** FK fields go through
895 /// [`Self::select_related`] (batched IN) or
896 /// [`Self::join_related`] (LEFT JOIN).
897 /// - **i64 parent PK only.** Same constraint as the rest of the
898 /// M2M plumbing; models with non-i64 PKs surface a clean
899 /// compile error.
900 /// - **Unknown field name → loud error** (post-#42). If the
901 /// name matches neither an M2M nor a `ReverseSet` field,
902 /// fetch returns a clear `sqlx::Error::Protocol` pointing at
903 /// the right method.
904 pub fn prefetch_related(mut self, field_name: impl Into<String>) -> Self {
905 self.prefetch_related.push(field_name.into());
906 self
907 }
908
909 /// Eagerly load multiple M2M relations. Sugar for chained
910 /// `.prefetch_related(name)` calls.
911 pub fn prefetch_related_many(mut self, field_names: &[&str]) -> Self {
912 for name in field_names {
913 self.prefetch_related.push(name.to_string());
914 }
915 self
916 }
917
918 /// Convert this QuerySet into a [`Subquery`] suitable for use in
919 /// an `IN (SELECT ...)` predicate. Projects only the named
920 /// column; the accumulated WHERE / ORDER BY survive.
921 ///
922 /// `Post::objects().filter(...).into_subquery("author_id")` →
923 /// `Subquery` you can hand to `user::ID.in_subquery(...)`.
924 pub fn into_subquery(self, col_name: &str) -> crate::orm::Subquery {
925 let mut q = self.build_query_for("sqlite");
926 q.clear_selects();
927 q.column(Alias::new(col_name));
928 crate::orm::Subquery::from_select(q)
929 }
930
931 /// Combine this QuerySet with `other` via SQL `UNION` (gap #28).
932 /// Both QuerySets must produce the same column shape — which
933 /// they always do here because both are typed `QuerySet<T>`.
934 /// Duplicates are removed (the de-duplicating UNION, not UNION
935 /// ALL).
936 pub fn union(self, other: QuerySet<T>) -> Self {
937 self.combine(other, sea_query::UnionType::Distinct)
938 }
939
940 /// Combine this QuerySet with `other` via SQL `INTERSECT`
941 /// (gap #28). Returns rows present in BOTH inputs.
942 pub fn intersect(self, other: QuerySet<T>) -> Self {
943 self.combine(other, sea_query::UnionType::Intersect)
944 }
945
946 /// Combine this QuerySet with `other` via SQL `EXCEPT`
947 /// (gap #28). Returns rows present in `self` but not in `other`.
948 pub fn except(self, other: QuerySet<T>) -> Self {
949 self.combine(other, sea_query::UnionType::Except)
950 }
951
952 /// Internal: attach `other`'s SelectStatement to `self`'s
953 /// SelectStatement with the given UnionType. Both sides apply
954 /// their accumulated predicates / ORDER BY before the union.
955 fn combine(mut self, other: QuerySet<T>, ty: sea_query::UnionType) -> Self {
956 let backend = "sqlite";
957 let other_select = other.build_query_for(backend);
958 // Fold our own predicates into the base query so the union
959 // sees them; further `.filter()` calls on the returned
960 // QuerySet would still apply to the OUTER (combined) query.
961 let mut base = self.build_query_for(backend);
962 self.predicates.clear();
963 base.union(ty, other_select);
964 self.query = base;
965 self
966 }
967
968 /// Emit `SELECT DISTINCT ...` for this query (gap #17). Most
969 /// useful when combined with [`Self::values`] to dedupe a
970 /// column-projected list (`distinct().values(&["tag"])`); the
971 /// full-row DISTINCT is rarely what you want.
972 ///
973 /// Postgres-specific `DISTINCT ON (cols)` is deferred until a
974 /// real consumer surfaces the need — the standard `DISTINCT`
975 /// covers most use cases.
976 pub fn distinct(mut self) -> Self {
977 self.query.distinct();
978 self
979 }
980}
981
982/// Resolve the pool to run a terminal against.
983///
984/// Precedence: explicit `.on(&pool)` / `.on_pg(&pool)` override wins;
985/// then the per-model database alias the Plugin contract published
986/// via `Plugin::database()` (FEATURES.md #6); then the `"default"`
987/// pool. Tests that skip the App builder pass an explicit pool and
988/// bypass the alias lookup entirely.
989/// Validate that every name passed to `.join_related(...)` resolves
990/// to a foreign-key field on `T`. Loud-error path that replaces the
991/// pre-#42 silent no-op (where an unknown field, an M2M field, or a
992/// non-FK column produced an empty JOIN list and a confusing
993/// "ForeignKey::resolved() returned None" downstream).
994fn validate_join_related_fields<T: Model>(fields: &[String]) -> Result<(), sqlx::Error> {
995 for field_name in fields {
996 // Nested path (`"plugin__author"` / `"tags__category"`):
997 // validate via the hop resolver. A leading M2M segment is a
998 // valid chain entry even though it isn't an FK on `T`, so
999 // accept it and let `apply_join_related` route it.
1000 if field_name.contains("__") {
1001 let first = field_name.split("__").next().unwrap_or(field_name);
1002 let leads_m2m = T::M2M_RELATIONS.iter().any(|r| r.field_name == first);
1003 if leads_m2m || resolve_join_hops::<T>(field_name).is_some() {
1004 continue;
1005 }
1006 return Err(sqlx::Error::Protocol(format!(
1007 "umbral::orm::join_related: nested path `{field_name}` on `{}` has an \
1008 unresolvable hop (each segment must be a FK or M2M to a registered model)",
1009 T::NAME
1010 )));
1011 }
1012 // Try as a regular column first.
1013 let col = T::FIELDS.iter().find(|f| f.name == field_name.as_str());
1014 if let Some(col) = col {
1015 if col.fk_target.is_some() {
1016 continue; // FK column — OK.
1017 }
1018 return Err(sqlx::Error::Protocol(format!(
1019 "umbral::orm::join_related: field `{field_name}` on `{}` is not a foreign \
1020 key (it has no fk_target)",
1021 T::NAME
1022 )));
1023 }
1024 // M2M field name? Post-#113 these go through the double
1025 // LEFT JOIN path: apply_join_related emits
1026 // `LEFT JOIN <junction> LEFT JOIN <child>` with aliased
1027 // child cols, and fetch()'s dedup-aware path collects M2M
1028 // children per parent. Trade-off documented at the join_related
1029 // docstring — M2M JOINs multiply parent rows by avg
1030 // cardinality, so prefetch_related stays the default for
1031 // any list page where M2M cardinality isn't tiny.
1032 if T::M2M_RELATIONS
1033 .iter()
1034 .any(|r| r.field_name == field_name.as_str())
1035 {
1036 continue;
1037 }
1038 return Err(sqlx::Error::Protocol(format!(
1039 "umbral::orm::join_related: unknown field `{field_name}` on model `{}`",
1040 T::NAME
1041 )));
1042 }
1043 Ok(())
1044}
1045
1046/// One resolved hop of a `join_related` FK chain.
1047#[derive(Debug, Clone)]
1048pub(crate) struct JoinHop {
1049 /// FK column name on the *previous* level's table.
1050 pub(crate) fk_col: String,
1051 /// Table this hop targets.
1052 pub(crate) child_table: String,
1053 /// PK column on `child_table`.
1054 pub(crate) child_pk: String,
1055 /// Was the FK column nullable? (drives auto-inference)
1056 pub(crate) nullable: bool,
1057}
1058
1059/// Resolve a dotted FK path (`"plugin__author"`) into ordered hops.
1060/// Hop 0 reads `T::FIELDS`; deeper hops read the migrate registry's
1061/// `Column`s for the prior hop's target table. Returns `None` (skip,
1062/// emit no JOIN) on any unresolved hop — same forgiving posture as the
1063/// pre-existing one-hop path's silent skip in `to_sql`.
1064///
1065/// FK-only: a path whose FIRST segment is an M2M field is NOT handled
1066/// here (M2M chains route through `apply_join_related`'s M2M branch);
1067/// this returns `None` for such a path.
1068pub(crate) fn resolve_join_hops<T: Model>(path: &str) -> Option<Vec<JoinHop>> {
1069 let registered = crate::migrate::registered_models();
1070 let segs: Vec<&str> = path.split("__").filter(|s| !s.is_empty()).collect();
1071 if segs.is_empty() {
1072 return None;
1073 }
1074 let mut hops = Vec::with_capacity(segs.len());
1075 // Hop 0 off the typed parent.
1076 let f0 = T::FIELDS.iter().find(|f| f.name == segs[0])?;
1077 let t0 = f0.fk_target?;
1078 let m0 = registered.iter().find(|m| m.table == t0)?;
1079 let pk0 = m0.fields.iter().find(|c| c.primary_key)?;
1080 hops.push(JoinHop {
1081 fk_col: segs[0].to_string(),
1082 child_table: t0.to_string(),
1083 child_pk: pk0.name.clone(),
1084 nullable: f0.nullable,
1085 });
1086 let mut current = t0;
1087 for seg in &segs[1..] {
1088 let meta = registered.iter().find(|m| m.table == current)?;
1089 let col = meta.fields.iter().find(|c| c.name == *seg)?;
1090 let tgt = col.fk_target.as_deref()?;
1091 let tmeta = registered.iter().find(|m| m.table == tgt)?;
1092 let pk = tmeta.fields.iter().find(|c| c.primary_key)?;
1093 hops.push(JoinHop {
1094 fk_col: (*seg).to_string(),
1095 child_table: tgt.to_string(),
1096 child_pk: pk.name.clone(),
1097 nullable: col.nullable,
1098 });
1099 current = tgt;
1100 }
1101 Some(hops)
1102}
1103
1104/// Thin wrapper so the backend hydration helpers (a sibling module)
1105/// can resolve a chain without importing the private name directly.
1106pub(crate) fn resolve_join_hops_for<T: Model>(path: &str) -> Option<Vec<JoinHop>> {
1107 resolve_join_hops::<T>(path)
1108}
1109
1110/// Resolve a path whose FIRST segment is an M2M field on `T` into the
1111/// M2M child table + child PK + the onward FK chain hops off that
1112/// child. `onward` is empty for a bare `"tags"`; for `"tags__category"`
1113/// it carries the `category` FK hop off the child table. `None` when
1114/// `segs[0]` isn't an M2M field or any onward hop fails to resolve.
1115pub(crate) fn resolve_m2m_chain<T: Model>(path: &str) -> Option<(String, String, Vec<JoinHop>)> {
1116 let registered = crate::migrate::registered_models();
1117 let segs: Vec<&str> = path.split("__").filter(|s| !s.is_empty()).collect();
1118 let first = segs.first()?;
1119 let rel = T::M2M_RELATIONS.iter().find(|r| r.field_name == *first)?;
1120 let child_meta = registered.iter().find(|m| m.table == rel.target_table)?;
1121 let child_pk = child_meta.fields.iter().find(|c| c.primary_key)?;
1122 let mut onward = Vec::with_capacity(segs.len().saturating_sub(1));
1123 let mut current = rel.target_table;
1124 for seg in &segs[1..] {
1125 let meta = registered.iter().find(|m| m.table == current)?;
1126 let col = meta.fields.iter().find(|c| c.name == *seg)?;
1127 let tgt = col.fk_target.as_deref()?;
1128 let tmeta = registered.iter().find(|m| m.table == tgt)?;
1129 let pk = tmeta.fields.iter().find(|c| c.primary_key)?;
1130 onward.push(JoinHop {
1131 fk_col: (*seg).to_string(),
1132 child_table: tgt.to_string(),
1133 child_pk: pk.name.clone(),
1134 nullable: col.nullable,
1135 });
1136 current = tgt;
1137 }
1138 Some((rel.target_table.to_string(), child_pk.name.clone(), onward))
1139}
1140
1141/// Gap #111 — error returned when a typed terminal (`fetch` / `first`
1142/// / `get`) runs against a QuerySet that has `.only(...)` set. A
1143/// partial-column row can't satisfy `T`'s `FromRow` impl, so the
1144/// caller has to either drop the `.only(...)` (full SELECT, typed
1145/// rows back) or terminate via `.values(&[...])` (JSON rows with
1146/// just the requested columns). The message names the offending
1147/// terminal so the fix is one rename away.
1148fn only_with_typed_terminal_error(terminal: &'static str) -> sqlx::Error {
1149 sqlx::Error::Protocol(format!(
1150 "umbral::orm::{terminal}: cannot run a typed terminal on a QuerySet \
1151 with `.only(...)` set — a partial-column row can't hydrate `T` via \
1152 FromRow. Either drop `.only(...)` to fetch full typed rows, or \
1153 terminate via `.values(&[...])` to get JSON rows with just the \
1154 projected columns."
1155 ))
1156}
1157
1158fn resolve_pool<T: Model>(explicit: Option<DbPool>, op: crate::db::RouteOp) -> DbPool {
1159 if let Some(pool) = explicit {
1160 return pool;
1161 }
1162 // Route through the swappable router when the registry is up.
1163 if let Some(meta) = crate::migrate::model_meta_ref(T::NAME) {
1164 let ctx = crate::db::route_context::current();
1165 let r = crate::db::router::router();
1166 let alias = match op {
1167 crate::db::RouteOp::Read => r.db_for_read(meta, &ctx),
1168 crate::db::RouteOp::Write => r.db_for_write(meta, &ctx),
1169 };
1170 return crate::db::pool_for_dispatched(alias.as_str()).clone();
1171 }
1172 // Registry-less fallback (low-level tests): today's static behavior.
1173 if let Some(alias) = crate::migrate::model_alias(T::NAME) {
1174 return crate::db::pool_for_dispatched(&alias).clone();
1175 }
1176 crate::db::pool_dispatched().clone()
1177}
1178
1179/// Pin a QuerySet to an explicit pool, dispatching SQLite vs Postgres. Used by
1180/// the upsert paths (`get_or_create` / `update_or_create`) so their
1181/// existence-check reads run on the WRITE database — read-your-writes, so a
1182/// read/write-split router never probes a lagging replica and inserts a
1183/// duplicate (or reads a stale row back after the update).
1184fn pin_to_pool<T: Model>(qs: QuerySet<T>, pool: &DbPool) -> QuerySet<T> {
1185 match pool {
1186 DbPool::Sqlite(p) => qs.on(p),
1187 DbPool::Postgres(p) => qs.on_pg(p),
1188 }
1189}
1190
1191// GetError / TryForEachError moved to `errors`; re-exported above.
1192
1193/// Emit a one-shot advisory when a `right_join_related` is applied
1194/// against a SQLite pool.
1195///
1196/// RIGHT/FULL JOIN landed in SQLite 3.39 (June 2022); Postgres has
1197/// always supported it. The boot system check (`check.rs`) can't surface
1198/// this — it's synchronous, has no live pool, and whether a RIGHT join
1199/// is *reachable* is a runtime QuerySet fact rather than static model
1200/// metadata. So the spec's "boot warning" is realized here: the first
1201/// time a RIGHT join is built against a SQLite pool, we `tracing::warn!`
1202/// once per process. We do NOT probe the library version (that needs an
1203/// async round-trip the SQL builder doesn't have) — the precise gate is
1204/// the SQLite driver's own error at execute time on an engine < 3.39;
1205/// this warn is the early nudge, consistent with `check.rs`'s
1206/// `Severity::Warning` posture.
1207///
1208/// Postgres pools and the no-pool case (a pure `to_sql` build with no
1209/// app booted) are silent.
1210fn warn_right_join_on_sqlite() {
1211 use std::sync::Once;
1212 static ONCE: Once = Once::new();
1213 if matches!(
1214 crate::db::try_pool_dispatched(),
1215 Some(crate::db::DbPool::Sqlite(_))
1216 ) {
1217 ONCE.call_once(|| {
1218 tracing::warn!(
1219 "umbral::orm::right_join_related: RIGHT JOIN requires SQLite >= 3.39. \
1220 If your SQLite is older the query will error at execute time; \
1221 Postgres is unaffected. Prefer left_/inner_join_related on SQLite \
1222 unless you've confirmed the engine version."
1223 );
1224 });
1225 }
1226}
1227
1228/// Terminal methods for every `QuerySet<T>` where `T: Model`.
1229///
1230/// Each terminal that materializes `T` carries a FromRow bound on the
1231/// method (not the impl block) — the conjunction of both backends'
1232/// FromRow impls. `#[derive(sqlx::FromRow)]` emits a generic-over-`R`
1233/// impl, so any user struct with standard field types satisfies both
1234/// bounds automatically.
1235impl<T: Model> QuerySet<T> {
1236 /// Render the SQL the QuerySet would execute, without running it.
1237 ///
1238 /// Returns the prepared statement with `?` placeholders for the
1239 /// bound values, exactly the string sqlx would send. Useful for
1240 /// `eprintln!`-style debugging and for tests that want to pin
1241 /// the rendered query without round-tripping through a pool.
1242 ///
1243 /// The bound values are intentionally not surfaced (sqlx's binder
1244 /// types aren't part of umbral's public surface); a `(sql, values)`
1245 /// accessor lands when EXPLAIN-style integration needs it.
1246 ///
1247 /// The rendered placeholder dialect is SQLite's (`?`). When the
1248 /// dispatched pool is Postgres the actual at-execute rendering
1249 /// uses `$1`-style placeholders; the `to_sql` debug surface
1250 /// continues to emit SQLite-style for stability across calls
1251 /// regardless of which pool is registered.
1252 pub fn to_sql(&self) -> String {
1253 let mut q = self.build_query_for("sqlite");
1254 self.apply_join_related(&mut q);
1255 self.apply_only_projection(&mut q);
1256 let (sql, _values) = q.build_sqlx(SqliteQueryBuilder);
1257 sql
1258 }
1259
1260 /// Render the QuerySet's SQL against the **Postgres** dialect,
1261 /// without running it. Companion to [`Self::to_sql`].
1262 ///
1263 /// The two render slightly different placeholder syntax (`?` for
1264 /// SQLite, `$1..$N` for Postgres) and any Postgres-specific
1265 /// operators like the array `@>` / `<@` / `&&` family only render
1266 /// correctly through this entry point — `to_sql`'s SQLite path
1267 /// leaves `$N` tokens in the template untouched. Use this when
1268 /// debugging a Postgres query or asserting on the rendered shape
1269 /// in tests.
1270 pub fn to_sql_pg(&self) -> String {
1271 let mut q = self.build_query_for("postgres");
1272 self.apply_join_related(&mut q);
1273 self.apply_only_projection(&mut q);
1274 let (sql, _values) = q.build_sqlx(PostgresQueryBuilder);
1275 sql
1276 }
1277
1278 /// Internal helper — when `.only(...)` was set, swap the SELECT
1279 /// list for just those columns. Shared by `to_sql` / `to_sql_pg`
1280 /// so the inspection surface stays in sync with what `values()`
1281 /// would emit. No-op when `only_cols` is `None`.
1282 fn apply_only_projection(&self, q: &mut sea_query::SelectStatement) {
1283 if let Some(cols) = &self.only_cols {
1284 q.clear_selects();
1285 for c in cols {
1286 q.column(Alias::new(c.as_str()));
1287 }
1288 }
1289 }
1290
1291 /// Internal helper — when `.join_related(name)` was set, wrap the
1292 /// current query as a subquery and build an outer SELECT that
1293 /// LEFT JOINs every requested related table (with child columns
1294 /// aliased as `<field>__<col>`). The subquery wrapper is the
1295 /// load-bearing trick: WHERE / ORDER BY / LIMIT predicates inside
1296 /// the inner query reference parent columns only — there's no
1297 /// JOIN in scope, so bare names like `id` resolve unambiguously
1298 /// to the parent. Without this wrapper SQLite raises
1299 /// "ambiguous column name: id" on any predicate sharing a column
1300 /// name with a JOIN'd table.
1301 ///
1302 /// No-op when `join_related` is empty. Unknown field names /
1303 /// unregistered related models / FK columns missing `fk_target`
1304 /// are silently skipped — the SQL just won't carry the JOIN and
1305 /// the caller notices when `ForeignKey::resolved()` stays empty.
1306 fn apply_join_related(&self, q: &mut sea_query::SelectStatement) {
1307 if self.join_related.is_empty() {
1308 return;
1309 }
1310 use sea_query::{Expr, Query};
1311 let registered = crate::migrate::registered_models();
1312
1313 // Inner-subquery column trim. When `.only(...)` is also set,
1314 // the outer SELECT only references a subset of parent
1315 // columns (plus the JOIN'd child columns it gets through
1316 // its alias). The inner subquery only needs to expose:
1317 // - parent columns named in `.only(...)` (intersected
1318 // with T::FIELDS so the JOIN aliases like
1319 // `category__name` don't leak in here — those live on
1320 // the JOIN'd table, not on the parent),
1321 // - PLUS the FK columns each `.join_related(name)` needs
1322 // for its `ON __p.<name> = <child>.<pk>` clause.
1323 // WHERE / ORDER BY inside the inner subquery can still
1324 // reference any parent column (SQL doesn't require an
1325 // ORDER BY column to be in the SELECT list), so we don't
1326 // need to promote those. Postgres often skips this prune
1327 // through subquery boundaries; SQLite usually doesn't —
1328 // either way trimming here is a measurable win on wide
1329 // tables (think 30-column Product on a busy hot path).
1330 if let Some(only) = &self.only_cols {
1331 let parent_field_names: std::collections::HashSet<&str> =
1332 T::FIELDS.iter().map(|f| f.name).collect();
1333 let mut needed: std::collections::HashSet<String> = only
1334 .iter()
1335 .filter(|c| parent_field_names.contains(c.as_str()))
1336 .cloned()
1337 .collect();
1338 for jr in &self.join_related {
1339 // Nested paths only need the FIRST hop's FK column at
1340 // the parent level; deeper hops join off the prior
1341 // level's alias, not the parent subquery.
1342 let join_field = jr.path.split("__").next().unwrap_or(jr.path.as_str());
1343 if parent_field_names.contains(join_field) {
1344 needed.insert(join_field.to_string());
1345 }
1346 }
1347 if !needed.is_empty() {
1348 q.clear_selects();
1349 // Stable ordering so the SQL is deterministic
1350 // across runs (HashSet iteration is not).
1351 let mut ordered: Vec<String> = needed.into_iter().collect();
1352 ordered.sort();
1353 for col in &ordered {
1354 q.column(Alias::new(col.as_str()));
1355 }
1356 }
1357 }
1358
1359 // Take ownership of the (possibly-trimmed) inner query and
1360 // re-mount it as the FROM clause of the new outer SELECT.
1361 let inner = std::mem::replace(q, Query::select().take());
1362 let parent_alias = Alias::new("__p");
1363 let mut outer = Query::select();
1364 outer.from_subquery(inner, parent_alias.clone());
1365 // Re-project the parent's full column set so the outer SELECT
1366 // exposes them unaliased — FromRow on `T` reads parent
1367 // columns by their bare names (`id`, `name`, ...). When
1368 // `.only(...)` later clears this list in
1369 // `apply_only_projection`, the inner-subquery trim above
1370 // means we still didn't pay for columns we ended up
1371 // dropping anyway.
1372 for f in T::FIELDS {
1373 outer.expr(Expr::col((parent_alias.clone(), Alias::new(f.name))));
1374 }
1375 // Set when any emitted hop is a RIGHT JOIN — drives the
1376 // once-per-process old-SQLite advisory after the emit loop.
1377 let mut emitted_right = false;
1378 for jr in &self.join_related {
1379 let field_name = &jr.path;
1380 // FK chain branch first. A (possibly nested) FK path splits
1381 // on `__` into ordered hops; each hop joins onto the prior
1382 // level's alias, and the DEEPEST hop's child columns are
1383 // aliased by the full dotted path so hydration can rebuild
1384 // the nested relation graph. The single-hop case is
1385 // byte-identical in child-column aliases to the pre-nesting
1386 // path (`<field>__<col>`); only the internal join alias
1387 // gains an `_h{idx}` suffix, which no test asserts.
1388 if let Some(hops) = resolve_join_hops::<T>(field_name) {
1389 let mut prev_alias = parent_alias.clone();
1390 let last = hops.len() - 1;
1391 // Cumulative dotted prefix per hop so EVERY level's own
1392 // columns ride along, aliased by its path-so-far. Hop 0
1393 // of `plugin__author` is `plugin`, hop 1 is
1394 // `plugin__author`. Selecting every level (not just the
1395 // leaf) is what lets hydration rebuild a FULL nested
1396 // object — the intermediate `plugin` row needs its own
1397 // `id`/`name` to deserialise into `ForeignKey<Plugin>`
1398 // before `author` nests inside it.
1399 let segs: Vec<&str> = field_name.split("__").collect();
1400 for (idx, hop) in hops.iter().enumerate() {
1401 let hop_alias = Alias::new(format!("__j_{field_name}_h{idx}"));
1402 // Last hop: explicit request, else infer from THIS
1403 // hop's nullability. Intermediate hops always infer
1404 // per-hop (Django nests an INNER inside an outer
1405 // LEFT etc.) — an explicit kind only pins the leaf.
1406 let kind = if idx == last {
1407 jr.kind.unwrap_or(if hop.nullable {
1408 JoinKind::Left
1409 } else {
1410 JoinKind::Inner
1411 })
1412 } else if hop.nullable {
1413 JoinKind::Left
1414 } else {
1415 JoinKind::Inner
1416 };
1417 emitted_right |= kind == JoinKind::Right;
1418 outer.join_as(
1419 kind.sea(),
1420 crate::db::router::schema_qualified_table(hop.child_table.as_str()),
1421 hop_alias.clone(),
1422 Expr::col((prev_alias.clone(), Alias::new(hop.fk_col.as_str())))
1423 .equals((hop_alias.clone(), Alias::new(hop.child_pk.as_str()))),
1424 );
1425 if let Some(meta) = registered.iter().find(|m| m.table == hop.child_table) {
1426 // Cumulative dotted prefix for this hop's columns.
1427 let prefix = segs[..=idx].join("__");
1428 for col in &meta.fields {
1429 let alias = format!("{}__{}", prefix, col.name);
1430 outer.expr_as(
1431 Expr::col((hop_alias.clone(), Alias::new(col.name.as_str()))),
1432 Alias::new(alias),
1433 );
1434 }
1435 }
1436 prev_alias = hop_alias;
1437 }
1438 continue;
1439 }
1440
1441 // M2M branch (post-#113). Emit the double LEFT JOIN
1442 // through the junction table:
1443 // LEFT JOIN <junction> AS __jm_<field>
1444 // ON __p.<parent_pk> = __jm_<field>.parent_id
1445 // LEFT JOIN <child_table> AS __j_<field>
1446 // ON __jm_<field>.child_id = __j_<field>.<child_pk>
1447 // Aliased child cols use the same `<field>__<col>` shape
1448 // as the FK branch so the decode helper can be reused.
1449 // The M2M field is the FIRST segment of the path; a nested
1450 // path like `"tags__category"` passes THROUGH the M2M hop
1451 // and continues with an onward FK chain off the child.
1452 let m2m_seg = field_name.split("__").next().unwrap_or(field_name.as_str());
1453 if let Some(m2m_rel) = T::M2M_RELATIONS.iter().find(|r| r.field_name == m2m_seg)
1454 && let Some(parent_pk) = T::FIELDS.iter().find(|f| f.primary_key)
1455 && let Some(child_meta) =
1456 registered.iter().find(|m| m.table == m2m_rel.target_table)
1457 && let Some(child_pk) = child_meta.fields.iter().find(|c| c.primary_key)
1458 {
1459 // Junction table + aliases key off the M2M field name
1460 // (segs[0]), NOT the full dotted path.
1461 let junction_table = format!("{}_{}", T::TABLE, m2m_seg);
1462 let junction_alias = Alias::new(format!("__jm_{m2m_seg}"));
1463 let child_alias = Alias::new(format!("__j_{m2m_seg}"));
1464 // The junction hop stays LEFT so a parent with zero
1465 // junction rows isn't dropped by the join to the
1466 // junction table itself — the CHILD hop's kind is what
1467 // decides drop/keep. Plain `join_related` (kind None)
1468 // leaves the child LEFT too, preserving the shipped
1469 // double-LEFT-JOIN M2M behavior (a tag-less parent
1470 // survives with an empty M2M slot). An explicit
1471 // inner_join_related drops parents whose relation is
1472 // absent: the junction-LEFT miss yields a NULL child_id,
1473 // then the child INNER on NULL has no match -> the parent
1474 // is dropped, which is the INNER contract.
1475 let child_kind = jr.kind.unwrap_or(JoinKind::Left);
1476 emitted_right |= child_kind == JoinKind::Right;
1477 outer.join_as(
1478 sea_query::JoinType::LeftJoin,
1479 crate::db::router::schema_qualified_table(&junction_table),
1480 junction_alias.clone(),
1481 Expr::col((parent_alias.clone(), Alias::new(parent_pk.name)))
1482 .equals((junction_alias.clone(), Alias::new("parent_id"))),
1483 );
1484 outer.join_as(
1485 child_kind.sea(),
1486 crate::db::router::schema_qualified_table(m2m_rel.target_table),
1487 child_alias.clone(),
1488 Expr::col((junction_alias.clone(), Alias::new("child_id")))
1489 .equals((child_alias.clone(), Alias::new(child_pk.name.as_str()))),
1490 );
1491 // Child columns aliased by the M2M field name so the
1492 // M2M decode path (`<m2m_field>__<col>`) reads them.
1493 for col in &child_meta.fields {
1494 let alias = format!("{}__{}", m2m_seg, col.name);
1495 outer.expr_as(
1496 Expr::col((child_alias.clone(), Alias::new(col.name.as_str()))),
1497 Alias::new(alias),
1498 );
1499 }
1500 // Onward FK chain off the child (segs[1..]). Each hop
1501 // joins onto the prior level's alias and aliases its
1502 // columns by the cumulative dotted path
1503 // (`tags__category__name`) so the M2M decode path can
1504 // nest the onward object into each child row.
1505 if let Some((_child_table, _child_pk, onward)) = resolve_m2m_chain::<T>(field_name)
1506 {
1507 let segs: Vec<&str> = field_name.split("__").collect();
1508 let mut prev_alias = child_alias.clone();
1509 for (i, hop) in onward.iter().enumerate() {
1510 // segs index for this hop: segs[0] is the M2M
1511 // field, segs[1] is onward[0], etc.
1512 let seg_idx = i + 1;
1513 let hop_alias = Alias::new(format!("__j_{m2m_seg}_o{i}"));
1514 let kind = if hop.nullable {
1515 JoinKind::Left
1516 } else {
1517 JoinKind::Inner
1518 };
1519 outer.join_as(
1520 kind.sea(),
1521 crate::db::router::schema_qualified_table(hop.child_table.as_str()),
1522 hop_alias.clone(),
1523 Expr::col((prev_alias.clone(), Alias::new(hop.fk_col.as_str())))
1524 .equals((hop_alias.clone(), Alias::new(hop.child_pk.as_str()))),
1525 );
1526 if let Some(meta) = registered.iter().find(|m| m.table == hop.child_table) {
1527 let prefix = segs[..=seg_idx].join("__");
1528 for col in &meta.fields {
1529 let alias = format!("{}__{}", prefix, col.name);
1530 outer.expr_as(
1531 Expr::col((hop_alias.clone(), Alias::new(col.name.as_str()))),
1532 Alias::new(alias),
1533 );
1534 }
1535 }
1536 prev_alias = hop_alias;
1537 }
1538 }
1539 continue;
1540 }
1541 }
1542 // A RIGHT JOIN against SQLite needs >= 3.39; warn once per
1543 // process. Postgres / no-pool builds stay silent.
1544 if emitted_right {
1545 warn_right_join_on_sqlite();
1546 }
1547 *q = outer;
1548 }
1549
1550 /// Run the SELECT and return every matching row.
1551 ///
1552 /// If `.select_related(name)` was called, a follow-up batch query
1553 /// populates `ForeignKey<U>.resolved` for each named field before
1554 /// the rows are returned.
1555 pub async fn fetch(self) -> Result<Vec<T>, sqlx::Error>
1556 where
1557 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1558 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
1559 + HydrateRelated,
1560 {
1561 if self.only_cols.is_some() {
1562 return Err(only_with_typed_terminal_error("fetch"));
1563 }
1564 let sr_fields = self.select_related.clone();
1565 let prefetch_fields = self.prefetch_related.clone();
1566 let join_reqs = self.join_related.clone();
1567 let join_fields: Vec<String> = join_reqs.iter().map(|j| j.path.clone()).collect();
1568 // Validate join_related field names up front so a typo or an
1569 // M2M field name doesn't silently no-op (it used to render a
1570 // SELECT with no JOIN). Pre-#42 the failure mode was
1571 // `ForeignKey::resolved()` stays None and the caller debugs
1572 // the wrong thing. Now they get a typed error.
1573 validate_join_related_fields::<T>(&join_fields)?;
1574 // The turbofish on `query_as_with::<DB, _, _>` is load-bearing:
1575 // with both `sqlx-sqlite` and `sqlx-postgres` features on
1576 // sea-query-binder, `SqlxValues` implements `IntoArguments` for
1577 // both backends, so the compiler can't infer DB from the values
1578 // alone. Naming DB explicitly pins which `FromRow` bound is
1579 // checked.
1580 // Split join_fields into FK vs M2M groups up front. The
1581 // M2M branch needs parent dedup (one parent row per JOIN
1582 // combo would surface duplicate Ts to the caller); the FK
1583 // branch is one-to-one with rows.
1584 // A path whose FIRST segment is an M2M field routes to the M2M
1585 // group even when it continues with an onward FK chain
1586 // (`"tags__category"`): the junction double-join + parent dedup
1587 // live on the M2M side, and the onward FK nests into each child.
1588 let (m2m_join_fields, fk_join_fields): (Vec<String>, Vec<String>) =
1589 join_fields.iter().cloned().partition(|f| {
1590 let first = f.split("__").next().unwrap_or(f.as_str());
1591 T::M2M_RELATIONS.iter().any(|r| r.field_name == first)
1592 });
1593 let has_m2m_join = !m2m_join_fields.is_empty();
1594
1595 let mut rows = match resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read)
1596 {
1597 DbPool::Sqlite(pool) => {
1598 let mut q = self.build_query_for("sqlite");
1599 self.apply_join_related(&mut q);
1600 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1601 if join_fields.is_empty() {
1602 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
1603 .fetch_all(&pool)
1604 .await?
1605 } else if !has_m2m_join {
1606 // FK-only JOIN path: one row in → one T out.
1607 let raw_rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
1608 .fetch_all(&pool)
1609 .await?;
1610 let mut typed = Vec::with_capacity(raw_rows.len());
1611 for row in &raw_rows {
1612 let mut t = <T as sqlx::FromRow<_>>::from_row(row)?;
1613 backend_sqlite::hydrate_joined_rels::<T>(&mut t, row, &fk_join_fields)?;
1614 typed.push(t);
1615 }
1616 typed
1617 } else {
1618 // Mixed (FK + M2M) or pure M2M JOIN path:
1619 // dedup parents, collect M2M children per
1620 // (parent_pk, field).
1621 let raw_rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
1622 .fetch_all(&pool)
1623 .await?;
1624 dedup_decode_sqlite::<T>(&raw_rows, &fk_join_fields, &m2m_join_fields)?
1625 }
1626 }
1627 DbPool::Postgres(pool) => {
1628 let mut q = self.build_query_for("postgres");
1629 self.apply_join_related(&mut q);
1630 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1631 if join_fields.is_empty() {
1632 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
1633 .fetch_all(&pool)
1634 .await?
1635 } else if !has_m2m_join {
1636 let raw_rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
1637 .fetch_all(&pool)
1638 .await?;
1639 let mut typed = Vec::with_capacity(raw_rows.len());
1640 for row in &raw_rows {
1641 let mut t = <T as sqlx::FromRow<_>>::from_row(row)?;
1642 backend_pg::hydrate_joined_rels::<T>(&mut t, row, &fk_join_fields)?;
1643 typed.push(t);
1644 }
1645 typed
1646 } else {
1647 let raw_rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
1648 .fetch_all(&pool)
1649 .await?;
1650 dedup_decode_pg::<T>(&raw_rows, &fk_join_fields, &m2m_join_fields)?
1651 }
1652 }
1653 };
1654 // BUG-16 step 2: wire each row's PK into its `M2M<U>` slots so
1655 // `add`/`remove`/`clear` know which parent they belong to.
1656 // No-op for models with no M2M fields.
1657 for r in &mut rows {
1658 r.set_m2m_parent_ids();
1659 }
1660 if !sr_fields.is_empty() {
1661 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
1662 hydrate_select_related::<T>(&mut rows, &sr_fields, &pool).await?;
1663 }
1664 if !prefetch_fields.is_empty() {
1665 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
1666 hydrate_prefetch_related::<T>(&mut rows, &prefetch_fields, &pool).await?;
1667 }
1668 Ok(rows)
1669 }
1670
1671 /// Feature 29 Phase 1 — chunked streaming via a callback.
1672 ///
1673 /// Runs the SELECT in pages of `chunk_size` rows and invokes
1674 /// `callback` once per row. Memory bound = `chunk_size *
1675 /// sizeof::<T>` instead of the full row count `fetch()` would
1676 /// buffer, so this is the right shape for million-row exports,
1677 /// migrations, and batch transforms.
1678 ///
1679 /// Deliberately NOT named `iterator()` — that name suggests a
1680 /// `Stream`-shaped return value, which would force a
1681 /// `futures-util` dep. The callback shape is idiomatic Rust,
1682 /// requires no new crates, and ships the same memory bound. A
1683 /// future `iterator()` returning `BoxStream<T>` can land later
1684 /// once `futures-util` is in the workspace for some other reason
1685 /// (likely SSE / WebSockets).
1686 ///
1687 /// Error contract: the callback may return any error type `E`.
1688 /// SQL failures become `TryForEachError::Sqlx`; callback errors
1689 /// become `TryForEachError::Callback(e)`. The first error stops
1690 /// the walk — subsequent rows are not fetched.
1691 ///
1692 /// Caveats: pages are stable only if the result set isn't being
1693 /// mutated concurrently. For consistent-snapshot iteration over
1694 /// a live table, wrap the call in a serialised-or-repeatable-read
1695 /// transaction. `select_related` and `prefetch_related` hooks
1696 /// are NOT applied on each row — `try_for_each` is intentionally
1697 /// the "raw column data, one row at a time" terminal.
1698 pub async fn try_for_each<F, E>(
1699 self,
1700 chunk_size: usize,
1701 mut callback: F,
1702 ) -> Result<(), TryForEachError<E>>
1703 where
1704 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1705 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
1706 + HydrateRelated,
1707 F: FnMut(T) -> Result<(), E>,
1708 {
1709 let chunk_size = chunk_size.max(1);
1710 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
1711 let mut offset: u64 = 0;
1712 loop {
1713 let mut rows: Vec<T> = match &pool {
1714 DbPool::Sqlite(pg) => {
1715 let mut q = self.build_query_for("sqlite");
1716 q.limit(chunk_size as u64).offset(offset);
1717 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1718 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
1719 .fetch_all(pg)
1720 .await
1721 .map_err(TryForEachError::Sqlx)?
1722 }
1723 DbPool::Postgres(pg) => {
1724 let mut q = self.build_query_for("postgres");
1725 q.limit(chunk_size as u64).offset(offset);
1726 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1727 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
1728 .fetch_all(pg)
1729 .await
1730 .map_err(TryForEachError::Sqlx)?
1731 }
1732 };
1733 let fetched = rows.len();
1734 if fetched == 0 {
1735 break;
1736 }
1737 for row in rows.drain(..) {
1738 callback(row).map_err(TryForEachError::Callback)?;
1739 }
1740 if fetched < chunk_size {
1741 break;
1742 }
1743 offset += fetched as u64;
1744 }
1745 Ok(())
1746 }
1747
1748 /// Run the SELECT with LIMIT 1 and return the first row, if any.
1749 pub async fn first(mut self) -> Result<Option<T>, sqlx::Error>
1750 where
1751 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1752 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
1753 + HydrateRelated,
1754 {
1755 if self.only_cols.is_some() {
1756 return Err(only_with_typed_terminal_error("first"));
1757 }
1758 // Review #2: delegate to `fetch()` with LIMIT 1 so select_related,
1759 // prefetch_related, AND join_related are all hydrated. `first()`
1760 // used to build a plain query and hydrate only select_related, so
1761 // `.prefetch_related("tags").first()` returned an unprefetched row
1762 // and `.join_related("author").first()` an unresolved join — both
1763 // silently. (For a to-many `join_related`, LIMIT 1 truncates the
1764 // joined children the same way `fetch()` does with `.limit(1)`;
1765 // prefer `prefetch_related` there.)
1766 self.query.limit(1);
1767 let rows = self.fetch().await?;
1768 Ok(rows.into_iter().next())
1769 }
1770
1771 /// Return the row with the smallest value in `col_name`. Sugar
1772 /// for `order_by(col.asc()).first()`. Mirrors Django's
1773 /// `QuerySet.earliest('created_at')`.
1774 ///
1775 /// Takes a `&'static str` column name (same shape as
1776 /// `select_related`) so the call site stays Django-flavoured —
1777 /// `.earliest("created_at")` reads naturally without spelling out
1778 /// `.asc()`.
1779 pub async fn earliest(self, col_name: &'static str) -> Result<Option<T>, sqlx::Error>
1780 where
1781 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1782 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
1783 + HydrateRelated,
1784 {
1785 self.order_by(OrderExpr::new(col_name, false)).first().await
1786 }
1787
1788 /// Return the row with the largest value in `col_name`. Sugar
1789 /// for `order_by(col.desc()).first()`. Mirrors Django's
1790 /// `QuerySet.latest('created_at')`.
1791 pub async fn latest(self, col_name: &'static str) -> Result<Option<T>, sqlx::Error>
1792 where
1793 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1794 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
1795 + HydrateRelated,
1796 {
1797 self.order_by(OrderExpr::new(col_name, true)).first().await
1798 }
1799
1800 /// Fetch many rows by their primary keys and return a
1801 /// `HashMap<T::PrimaryKey, T>` keyed by PK. The everyday companion to
1802 /// a cached list of ids — `User::objects().in_bulk(user_ids)` gives
1803 /// you direct lookup access without a second `.iter().find(...)` pass
1804 /// per id.
1805 ///
1806 /// Missing ids are silently absent from the map; callers that
1807 /// need the existence check can compare `map.len()` to
1808 /// `pks.len()`. Empty input is a no-op (returns the empty map).
1809 ///
1810 /// PK-agnostic (PK lift — was `Vec<i64>` / `HashMap<i64, T>`): the key
1811 /// is the model's `PrimaryKey` type, so i64-, String/slug-, and
1812 /// Uuid-keyed models all work. The map key requires `Hash + Eq`,
1813 /// which every standard PK type (integers, `String`, `Uuid`) satisfies.
1814 pub async fn in_bulk(
1815 self,
1816 pks: Vec<T::PrimaryKey>,
1817 ) -> Result<HashMap<T::PrimaryKey, T>, sqlx::Error>
1818 where
1819 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1820 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
1821 + HydrateRelated,
1822 T::PrimaryKey: std::hash::Hash + Eq,
1823 {
1824 if pks.is_empty() {
1825 return Ok(HashMap::new());
1826 }
1827 let pk_name = pk_field::<T>().map(|f| f.name).unwrap_or("id");
1828 // Each PK converts to a `sea_query::Value` (the `PrimaryKey` trait
1829 // bounds `Into<sea_query::Value>`); wrap as a `SimpleExpr` for the
1830 // IN-list so any PK shape binds correctly.
1831 let pk_pred: Predicate<T> = Predicate::new(
1832 Expr::col(Alias::new(pk_name)).is_in(
1833 pks.into_iter()
1834 .map(|p| sea_query::SimpleExpr::Value(p.into())),
1835 ),
1836 );
1837 let rows = self.filter(pk_pred).fetch().await?;
1838 let mut out: HashMap<T::PrimaryKey, T> = HashMap::with_capacity(rows.len());
1839 for row in rows {
1840 out.insert(row.primary_key(), row);
1841 }
1842 Ok(out)
1843 }
1844
1845 /// Return the database's execution plan for this query as a
1846 /// plain-text string. Doesn't run the underlying query — just
1847 /// asks the DB how it would be executed.
1848 ///
1849 /// Backend dispatch:
1850 ///
1851 /// - SQLite: `EXPLAIN QUERY PLAN <sql>` — returns the planner's
1852 /// nested loop hierarchy, one row per access step.
1853 /// - Postgres: `EXPLAIN <sql>` — returns the default text plan.
1854 /// For machine-readable output use raw sqlx with
1855 /// `EXPLAIN (FORMAT JSON)`; the framework defaults to text
1856 /// because most callers want eyeball-able output.
1857 ///
1858 /// Lines are joined with newlines. The returned string is what a
1859 /// developer would paste into a debugger or a perf-review issue.
1860 pub async fn explain(self) -> Result<String, sqlx::Error> {
1861 // Annotations are part of the built query (they render inside
1862 // build_query_for), so the plan below includes them — but a
1863 // poisoned annotation (unknown relation) must fail loudly
1864 // here, not silently vanish from the plan.
1865 self.check_annotations()?;
1866 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
1867 let backend = pool.backend_name();
1868 let q = self.build_query_for(backend);
1869 match pool {
1870 DbPool::Sqlite(pool) => {
1871 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
1872 let explain_sql = format!("EXPLAIN QUERY PLAN {sql}");
1873 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&explain_sql, vals)
1874 .fetch_all(&pool)
1875 .await?;
1876 let mut out = String::new();
1877 for row in &rows {
1878 use sqlx::Row;
1879 // SQLite returns: id, parent, notused, detail.
1880 // The `detail` column is the human-readable step.
1881 let detail: String = row.try_get("detail")?;
1882 if !out.is_empty() {
1883 out.push('\n');
1884 }
1885 out.push_str(&detail);
1886 }
1887 Ok(out)
1888 }
1889 DbPool::Postgres(pool) => {
1890 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1891 let explain_sql = format!("EXPLAIN {sql}");
1892 let rows = sqlx::query_with::<sqlx::Postgres, _>(&explain_sql, vals)
1893 .fetch_all(&pool)
1894 .await?;
1895 let mut out = String::new();
1896 for row in &rows {
1897 use sqlx::Row;
1898 // Postgres EXPLAIN returns one column named
1899 // "QUERY PLAN", one row per line of the plan.
1900 let line: String = row.try_get("QUERY PLAN")?;
1901 if !out.is_empty() {
1902 out.push('\n');
1903 }
1904 out.push_str(&line);
1905 }
1906 Ok(out)
1907 }
1908 }
1909 }
1910
1911 /// Run `SELECT COUNT(*)` against the same FROM + WHERE.
1912 ///
1913 /// Reshapes the query rather than wrapping the existing SELECT: the
1914 /// projection becomes `COUNT(*)` and LIMIT/OFFSET drop away. ORDER
1915 /// BY is harmless on a scalar aggregate and is left in place. The
1916 /// row type is `(i64,)` so the FromRow constraint comes from sqlx's
1917 /// tuple impl rather than the user struct — count() doesn't need
1918 /// T's FromRow bounds.
1919 pub async fn count(self) -> Result<i64, sqlx::Error> {
1920 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
1921 let backend = pool.backend_name();
1922 // Build the dialect-appropriate filtered query first, then
1923 // rebuild as COUNT. Doing it in this order keeps the predicate
1924 // walk pluggable per backend without duplicating the COUNT
1925 // rewrite logic across branches.
1926 let mut rebuilt = self.build_query_for(backend);
1927 rebuilt.clear_selects();
1928 // Postgres rejects `"*"` as a quoted identifier (SQLite tolerates
1929 // it); use sea_query's Asterisk token which renders bare `*`
1930 // on both backends.
1931 rebuilt.expr(Func::count(Expr::col(sea_query::Asterisk)));
1932 rebuilt.reset_limit();
1933 rebuilt.reset_offset();
1934
1935 match pool {
1936 DbPool::Sqlite(pool) => {
1937 let (sql, values) = rebuilt.build_sqlx(SqliteQueryBuilder);
1938 let (n,): (i64,) = sqlx::query_as_with::<sqlx::Sqlite, (i64,), _>(&sql, values)
1939 .fetch_one(&pool)
1940 .await?;
1941 Ok(n)
1942 }
1943 DbPool::Postgres(pool) => {
1944 let (sql, values) = rebuilt.build_sqlx(PostgresQueryBuilder);
1945 let (n,): (i64,) = sqlx::query_as_with::<sqlx::Postgres, (i64,), _>(&sql, values)
1946 .fetch_one(&pool)
1947 .await?;
1948 Ok(n)
1949 }
1950 }
1951 }
1952
1953 /// Return whether any row matches.
1954 ///
1955 /// M1 keeps the simple form: add LIMIT 1, fetch, check non-empty.
1956 /// A later milestone may swap the projection for `SELECT 1` to
1957 /// skip column materialisation.
1958 pub async fn exists(self) -> Result<bool, sqlx::Error>
1959 where
1960 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1961 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
1962 + HydrateRelated,
1963 {
1964 let rows = self.limit(1).fetch().await?;
1965 Ok(!rows.is_empty())
1966 }
1967
1968 /// `.get()` — Django's exactly-one terminal.
1969 ///
1970 /// Returns `Ok(row)` when the filter chain matches exactly one
1971 /// row. The two not-exactly-one cases each get their own
1972 /// `GetError` variant so the caller can branch deliberately:
1973 ///
1974 /// - [`GetError::NotFound`] — zero rows matched. The right
1975 /// choice for "fetch the row this user just clicked on; 404
1976 /// if it's gone."
1977 /// - [`GetError::MultipleObjectsReturned`] — more than one row
1978 /// matched. The right choice for filters that should be
1979 /// uniquely-keyed (e.g. `.filter(user::EMAIL.eq("..."))`
1980 /// when email has a UNIQUE constraint); a result of 2+ is a
1981 /// data-integrity bug worth crashing on.
1982 /// - The underlying sqlx error wraps as [`GetError::Sqlx`].
1983 ///
1984 /// Internally this issues `SELECT ... LIMIT 2` — the cheapest
1985 /// way to distinguish "one row" from "many." The second row, if
1986 /// it exists, isn't materialised beyond the bare FromRow call.
1987 ///
1988 /// ```ignore
1989 /// match Post::objects().filter(post::ID.eq(42)).get().await {
1990 /// Ok(p) => /* render */,
1991 /// Err(GetError::NotFound) => /* 404 */,
1992 /// Err(GetError::MultipleObjectsReturned) => unreachable!("ID is unique"),
1993 /// Err(GetError::Sqlx(e)) => /* 500 */,
1994 /// }
1995 /// ```
1996 pub async fn get(self) -> Result<T, GetError>
1997 where
1998 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1999 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
2000 + HydrateRelated,
2001 {
2002 let mut rows = self.limit(2).fetch().await.map_err(GetError::Sqlx)?;
2003 match rows.len() {
2004 0 => Err(GetError::NotFound),
2005 1 => Ok(rows.pop().unwrap()),
2006 _ => Err(GetError::MultipleObjectsReturned),
2007 }
2008 }
2009
2010 // =====================================================================
2011 // Postgres-only terminals (Phase 4.1).
2012 //
2013 // Models with Postgres-only field types (`Vec<T>` arrays, the future
2014 // Hstore / CIDR / FullTextSearch types) can't satisfy the dual
2015 // FromRow bound on `fetch` / `first` / `count` / `exists`. These
2016 // `_pg` variants bound on `FromRow<PgRow>` alone, take the pool as
2017 // an argument, and skip the dispatch — the call site explicitly
2018 // says "this model is Postgres-only."
2019 //
2020 // For models with portable fields, the existing `fetch` etc. stay
2021 // the recommended call: they pick up the ambient pool and route
2022 // through `.on(&pool)` / `.on_pg(&pool)` overrides exactly as
2023 // Phase 2.5 documented.
2024 // =====================================================================
2025
2026 // =====================================================================
2027 // Write terminals — DELETE and UPDATE.
2028 //
2029 // Both apply the accumulated filter predicates as the WHERE clause,
2030 // dispatch to the resolved pool's backend, and return the affected-
2031 // rows count from sqlx. No row materialisation — DELETE is keyless,
2032 // and UPDATE doesn't do a RETURNING read-back at v1 (use
2033 // `.filter(...).fetch()` after a write if you need the updated
2034 // rows back).
2035 //
2036 // **Without a `.filter(...)`, both terminals affect every row in
2037 // the table.** That mirrors raw SQL semantics; the type system
2038 // can't distinguish "I forgot the filter" from "I really meant to
2039 // truncate." Users protecting against accidental full-table writes
2040 // wrap their callers or assert a row count via `.count()` first.
2041 // =====================================================================
2042
2043 /// Project the query to only the named columns, returning a
2044 /// vector of `serde_json::Value::Object` rows instead of typed
2045 /// `T` instances. Mirrors Django's `QuerySet.values('id', 'title')`.
2046 ///
2047 /// Use when a list view only needs a few fields — skipping the
2048 /// 50KB body BLOB on every Post saves both memory and the
2049 /// FromRow hydration overhead. Each returned `Value` is an
2050 /// object keyed by the requested column names, with values
2051 /// typed per the column's declared SqlType (integers stay
2052 /// integers, booleans stay booleans, dates render as ISO
2053 /// strings).
2054 ///
2055 /// Unknown column names fail loudly with
2056 /// `sqlx::Error::Protocol` naming the offending column.
2057 /// Composes with `filter`, `exclude`, `order_by`, `limit`,
2058 /// `offset` exactly the way the typed terminals do.
2059 ///
2060 /// ```rust,ignore
2061 /// let rows = Post::objects()
2062 /// .filter(post::PUBLISHED.eq(true))
2063 /// .order_by(post::ID.desc())
2064 /// .values(&["id", "title"])
2065 /// .await?;
2066 /// // [ { "id": 3, "title": "c" }, ... ]
2067 /// ```
2068 pub async fn values(self, columns: &[&str]) -> Result<Vec<JsonValue>, sqlx::Error> {
2069 // Gap #46 follow-up: if any name uses `__` traversal
2070 // (`author__id`), route to the JOIN-aware path that builds
2071 // nested per-relation JSON objects. The unbranched path
2072 // below stays byte-for-byte identical for the common
2073 // parent-cols-only case.
2074 if columns.iter().any(|c| c.contains("__")) {
2075 return self.values_with_traversal(columns).await;
2076 }
2077 let meta = crate::migrate::ModelMeta::for_::<T>();
2078 // Resolve every requested name against the model's metadata
2079 // up front so an unknown column errors before any SQL runs.
2080 let mut chosen: Vec<&crate::migrate::Column> = Vec::with_capacity(columns.len());
2081 for name in columns {
2082 let col = meta
2083 .fields
2084 .iter()
2085 .find(|c| c.name == *name)
2086 .ok_or_else(|| {
2087 sqlx::Error::Protocol(format!(
2088 "umbral::orm::values: unknown column `{}` on model `{}`",
2089 name,
2090 T::NAME
2091 ))
2092 })?;
2093 chosen.push(col);
2094 }
2095 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2096 let backend = pool.backend_name();
2097 // Build the base query (predicates + ORDER BY) then swap its
2098 // SELECT list for only the requested columns.
2099 let mut q = self.build_query_for(backend);
2100 q.clear_selects();
2101 for col in &chosen {
2102 q.column(Alias::new(col.name.as_str()));
2103 }
2104 match pool {
2105 DbPool::Sqlite(pool) => {
2106 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
2107 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, vals)
2108 .fetch_all(&pool)
2109 .await?;
2110 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2111 for row in &rows {
2112 let mut obj = serde_json::Map::with_capacity(chosen.len());
2113 for col in &chosen {
2114 let v = crate::orm::dynamic::decode_to_json(row, col)?;
2115 obj.insert(col.name.clone(), v);
2116 }
2117 out.push(JsonValue::Object(obj));
2118 }
2119 Ok(out)
2120 }
2121 DbPool::Postgres(pool) => {
2122 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
2123 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, vals)
2124 .fetch_all(&pool)
2125 .await?;
2126 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2127 for row in &rows {
2128 let mut obj = serde_json::Map::with_capacity(chosen.len());
2129 for col in &chosen {
2130 let v = crate::orm::dynamic::decode_pg_to_json(row, col)?;
2131 obj.insert(col.name.clone(), v);
2132 }
2133 out.push(JsonValue::Object(obj));
2134 }
2135 Ok(out)
2136 }
2137 }
2138 }
2139
2140 /// `.values("author__name")`-style traversal. One-hop only at
2141 /// v1 (`a__b`, not `a__b__c` — that fails loudly so the user
2142 /// doesn't get a silent partial result). Emits one LEFT JOIN
2143 /// per distinct relation referenced, aliases child columns as
2144 /// `<rel>__<col>`, and returns each row as a nested JSON
2145 /// object: `{id, title, author: {id, name}, editor: {id}}`.
2146 ///
2147 /// A LEFT JOIN miss (nullable FK pointing at nothing) maps the
2148 /// whole relation key to `Value::Null` rather than a nested
2149 /// object full of nulls — caller code that branches on
2150 /// `obj["author"].is_null()` works naturally.
2151 ///
2152 /// Validates every name up front so a typo errors before any
2153 /// SQL runs. Unknown parent col / non-FK relation name /
2154 /// unknown child col / deeper-than-one-hop path all surface
2155 /// distinct messages.
2156 async fn values_with_traversal(self, columns: &[&str]) -> Result<Vec<JsonValue>, sqlx::Error> {
2157 use sea_query::{Expr, Query};
2158 let meta = crate::migrate::ModelMeta::for_::<T>();
2159 let registered = crate::migrate::registered_models();
2160
2161 // Split each column name into (relation, child_col) or
2162 // (None, parent_col). Reject paths with more than one `__`
2163 // hop — nested traversal across two relation layers is a
2164 // separate piece of work (it'd need to chain JOINs and
2165 // build doubly-nested JSON).
2166 let mut parent_cols: Vec<String> = Vec::new();
2167 let mut per_rel: std::collections::BTreeMap<String, Vec<String>> =
2168 std::collections::BTreeMap::new();
2169 for raw in columns {
2170 let mut parts = raw.splitn(3, "__");
2171 let first = parts.next().unwrap_or("");
2172 let second = parts.next();
2173 let third = parts.next();
2174 if third.is_some() {
2175 return Err(sqlx::Error::Protocol(format!(
2176 "umbral::orm::values: nested `{raw}` is not supported in v1 \
2177 (one-hop only — `a__b`, not `a__b__c`)"
2178 )));
2179 }
2180 match second {
2181 Some(child) => {
2182 per_rel
2183 .entry(first.to_string())
2184 .or_default()
2185 .push(child.to_string());
2186 }
2187 None => parent_cols.push(first.to_string()),
2188 }
2189 }
2190
2191 // Validate every parent name against T::FIELDS.
2192 for name in &parent_cols {
2193 if !meta.fields.iter().any(|c| c.name == *name) {
2194 return Err(sqlx::Error::Protocol(format!(
2195 "umbral::orm::values: unknown column `{name}` on model `{}`",
2196 T::NAME
2197 )));
2198 }
2199 }
2200
2201 // Validate every relation + child trio. Build a struct per
2202 // relation that the SQL/decoder loops below need.
2203 struct RelInfo<'a> {
2204 rel_name: String,
2205 related_table: &'a str,
2206 related_pk: &'a crate::migrate::Column,
2207 child_cols: Vec<&'a crate::migrate::Column>,
2208 }
2209 let mut rel_infos: Vec<RelInfo<'_>> = Vec::with_capacity(per_rel.len());
2210 for (rel_name, child_names) in &per_rel {
2211 let fk_field = meta.fields.iter().find(|c| c.name == *rel_name);
2212 let Some(fk_field) = fk_field else {
2213 return Err(sqlx::Error::Protocol(format!(
2214 "umbral::orm::values: unknown relation `{rel_name}` on model `{}` \
2215 (used in `{rel_name}__...`)",
2216 T::NAME
2217 )));
2218 };
2219 let Some(related_table) = fk_field.fk_target.as_deref() else {
2220 return Err(sqlx::Error::Protocol(format!(
2221 "umbral::orm::values: field `{rel_name}` on `{}` is not a foreign \
2222 key — `__` traversal only works through FK fields",
2223 T::NAME
2224 )));
2225 };
2226 let Some(related_meta) = registered.iter().find(|m| m.table == related_table) else {
2227 return Err(sqlx::Error::Protocol(format!(
2228 "umbral::orm::values: related model for table `{related_table}` \
2229 is not registered"
2230 )));
2231 };
2232 let Some(related_pk) = related_meta.fields.iter().find(|c| c.primary_key) else {
2233 return Err(sqlx::Error::Protocol(format!(
2234 "umbral::orm::values: related model `{related_table}` has no \
2235 primary key column"
2236 )));
2237 };
2238 let mut child_cols: Vec<&crate::migrate::Column> =
2239 Vec::with_capacity(child_names.len());
2240 for name in child_names {
2241 let col = related_meta
2242 .fields
2243 .iter()
2244 .find(|c| c.name == *name)
2245 .ok_or_else(|| {
2246 sqlx::Error::Protocol(format!(
2247 "umbral::orm::values: unknown child column `{name}` on \
2248 related model `{related_table}` (full path `{rel_name}__{name}`)"
2249 ))
2250 })?;
2251 child_cols.push(col);
2252 }
2253 rel_infos.push(RelInfo {
2254 rel_name: rel_name.clone(),
2255 related_table,
2256 related_pk,
2257 child_cols,
2258 });
2259 }
2260
2261 // Build the SQL. Subquery-wrap the parent (WHERE / ORDER
2262 // BY / LIMIT all stay scoped to it) so JOIN'd tables can't
2263 // shadow bare-column predicates — same trick
2264 // apply_join_related uses.
2265 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2266 let backend = pool.backend_name();
2267 let inner = self.build_query_for(backend);
2268 let parent_alias = Alias::new("__p");
2269 let mut outer = Query::select();
2270 outer.from_subquery(inner, parent_alias.clone());
2271 // Outer SELECT: parent cols (bare), aliased child cols.
2272 for name in &parent_cols {
2273 outer.expr_as(
2274 Expr::col((parent_alias.clone(), Alias::new(name.as_str()))),
2275 Alias::new(name.as_str()),
2276 );
2277 }
2278 for info in &rel_infos {
2279 let join_alias = Alias::new(format!("__j_{}", info.rel_name));
2280 outer.join_as(
2281 sea_query::JoinType::LeftJoin,
2282 crate::db::router::schema_qualified_table(info.related_table),
2283 join_alias.clone(),
2284 Expr::col((parent_alias.clone(), Alias::new(info.rel_name.as_str()))).equals((
2285 join_alias.clone(),
2286 Alias::new(info.related_pk.name.as_str()),
2287 )),
2288 );
2289 // Always include the related PK alias so the decoder
2290 // can detect a LEFT JOIN miss → emit Value::Null for
2291 // the whole relation. Plus every requested child col.
2292 let pk_alias = format!("{}__{}", info.rel_name, info.related_pk.name);
2293 outer.expr_as(
2294 Expr::col((
2295 join_alias.clone(),
2296 Alias::new(info.related_pk.name.as_str()),
2297 )),
2298 Alias::new(pk_alias),
2299 );
2300 for col in &info.child_cols {
2301 let alias = format!("{}__{}", info.rel_name, col.name);
2302 outer.expr_as(
2303 Expr::col((join_alias.clone(), Alias::new(col.name.as_str()))),
2304 Alias::new(alias),
2305 );
2306 }
2307 }
2308
2309 // Execute + decode. Per-backend dispatch.
2310 match pool {
2311 DbPool::Sqlite(pool) => {
2312 let (sql, vals) = outer.build_sqlx(SqliteQueryBuilder);
2313 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, vals)
2314 .fetch_all(&pool)
2315 .await?;
2316 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2317 for row in &rows {
2318 let mut obj = serde_json::Map::new();
2319 // Parent cols decode by their bare alias name.
2320 for name in &parent_cols {
2321 if let Some(col) = meta.fields.iter().find(|c| c.name == *name) {
2322 let v = crate::orm::dynamic::decode_to_json_aliased(row, col, name)?;
2323 obj.insert(name.clone(), v);
2324 }
2325 }
2326 // Per-relation nested object — `null` on LEFT JOIN miss.
2327 for info in &rel_infos {
2328 let pk_alias = format!("{}__{}", info.rel_name, info.related_pk.name);
2329 let pk_is_null =
2330 backend_sqlite::joined_pk_is_null(row, &info.related_pk, &pk_alias);
2331 if pk_is_null {
2332 obj.insert(info.rel_name.clone(), JsonValue::Null);
2333 continue;
2334 }
2335 let mut nested = serde_json::Map::with_capacity(info.child_cols.len());
2336 for col in &info.child_cols {
2337 let alias = format!("{}__{}", info.rel_name, col.name);
2338 let v = crate::orm::dynamic::decode_to_json_aliased(row, col, &alias)?;
2339 nested.insert(col.name.clone(), v);
2340 }
2341 obj.insert(info.rel_name.clone(), JsonValue::Object(nested));
2342 }
2343 out.push(JsonValue::Object(obj));
2344 }
2345 Ok(out)
2346 }
2347 DbPool::Postgres(pool) => {
2348 let (sql, vals) = outer.build_sqlx(PostgresQueryBuilder);
2349 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, vals)
2350 .fetch_all(&pool)
2351 .await?;
2352 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2353 for row in &rows {
2354 let mut obj = serde_json::Map::new();
2355 for name in &parent_cols {
2356 if let Some(col) = meta.fields.iter().find(|c| c.name == *name) {
2357 let v = crate::orm::dynamic::decode_pg_to_json_aliased(row, col, name)?;
2358 obj.insert(name.clone(), v);
2359 }
2360 }
2361 for info in &rel_infos {
2362 let pk_alias = format!("{}__{}", info.rel_name, info.related_pk.name);
2363 let pk_is_null =
2364 backend_pg::joined_pk_is_null(row, &info.related_pk, &pk_alias);
2365 if pk_is_null {
2366 obj.insert(info.rel_name.clone(), JsonValue::Null);
2367 continue;
2368 }
2369 let mut nested = serde_json::Map::with_capacity(info.child_cols.len());
2370 for col in &info.child_cols {
2371 let alias = format!("{}__{}", info.rel_name, col.name);
2372 let v =
2373 crate::orm::dynamic::decode_pg_to_json_aliased(row, col, &alias)?;
2374 nested.insert(col.name.clone(), v);
2375 }
2376 obj.insert(info.rel_name.clone(), JsonValue::Object(nested));
2377 }
2378 out.push(JsonValue::Object(obj));
2379 }
2380 Ok(out)
2381 }
2382 }
2383 }
2384
2385 /// Single-row aggregate. Runs `SELECT AGG(col) AS name, ...` with
2386 /// the QuerySet's accumulated WHERE clause (ORDER BY / LIMIT /
2387 /// OFFSET are dropped — they make no sense over an aggregate
2388 /// without GROUP BY).
2389 ///
2390 /// Returns a `serde_json::Value::Object` keyed by the supplied
2391 /// names. COUNT comes back as an integer; AVG as a float; SUM /
2392 /// MAX / MIN inherit the source column's declared type.
2393 ///
2394 /// ```rust,ignore
2395 /// use umbral::orm::Aggregate;
2396 /// let summary = Post::objects()
2397 /// .filter(post::PUBLISHED.eq(true))
2398 /// .aggregate(&[
2399 /// ("count", Aggregate::count()),
2400 /// ("total", Aggregate::sum("view_count")),
2401 /// ])
2402 /// .await?;
2403 /// // { "count": 42, "total": 9999 }
2404 /// ```
2405 pub async fn aggregate(
2406 self,
2407 aggs: &[(&str, crate::orm::Aggregate)],
2408 ) -> Result<JsonValue, sqlx::Error> {
2409 let meta = crate::migrate::ModelMeta::for_::<T>();
2410 // Validate every aggregate's source column exists.
2411 for (name, agg) in aggs {
2412 if let Some(col) = agg.source_column()
2413 && !meta.fields.iter().any(|c| c.name == col)
2414 {
2415 return Err(sqlx::Error::Protocol(format!(
2416 "umbral::orm::aggregate: unknown column `{}` on model `{}` for aggregate `{}`",
2417 col,
2418 T::NAME,
2419 name
2420 )));
2421 }
2422 }
2423 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2424 let backend = pool.backend_name();
2425 let mut q = self.build_query_for(backend);
2426 q.clear_selects();
2427 q.reset_limit();
2428 q.reset_offset();
2429 for (name, agg) in aggs {
2430 q.expr_as(agg.to_simple_expr(), Alias::new(*name));
2431 }
2432 match pool {
2433 DbPool::Sqlite(pool) => {
2434 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
2435 let row = sqlx::query_with::<sqlx::Sqlite, _>(&sql, vals)
2436 .fetch_one(&pool)
2437 .await?;
2438 let mut obj = serde_json::Map::with_capacity(aggs.len());
2439 for (name, agg) in aggs {
2440 let source_ty = agg
2441 .source_column()
2442 .and_then(|c| meta.fields.iter().find(|f| f.name == c).map(|f| f.ty));
2443 obj.insert(
2444 name.to_string(),
2445 backend_sqlite::decode_agg(&row, name, agg, source_ty)?,
2446 );
2447 }
2448 Ok(JsonValue::Object(obj))
2449 }
2450 DbPool::Postgres(pool) => {
2451 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
2452 let row = sqlx::query_with::<sqlx::Postgres, _>(&sql, vals)
2453 .fetch_one(&pool)
2454 .await?;
2455 let mut obj = serde_json::Map::with_capacity(aggs.len());
2456 for (name, agg) in aggs {
2457 let source_ty = agg
2458 .source_column()
2459 .and_then(|c| meta.fields.iter().find(|f| f.name == c).map(|f| f.ty));
2460 obj.insert(
2461 name.to_string(),
2462 backend_pg::decode_agg(&row, name, agg, source_ty)?,
2463 );
2464 }
2465 Ok(JsonValue::Object(obj))
2466 }
2467 }
2468 }
2469
2470 /// Grouped aggregate. Runs `SELECT <group_cols>, AGG(col) AS name,
2471 /// ... GROUP BY <group_cols>` with the accumulated WHERE clause.
2472 ///
2473 /// Returns one `Value::Object` per group, with both the group
2474 /// columns and each named aggregate as fields. Group columns are
2475 /// decoded per their declared SqlType (so an integer
2476 /// `author_id` stays a JSON number).
2477 ///
2478 /// ```rust,ignore
2479 /// let by_author = Post::objects()
2480 /// .annotate(&["author_id"], &[("count", Aggregate::count())])
2481 /// .await?;
2482 /// // [ { "author_id": 1, "count": 3 }, { "author_id": 2, "count": 2 } ]
2483 /// ```
2484 pub async fn annotate(
2485 self,
2486 group_cols: &[&str],
2487 aggs: &[(&str, crate::orm::Aggregate)],
2488 ) -> Result<Vec<JsonValue>, sqlx::Error> {
2489 let meta = crate::migrate::ModelMeta::for_::<T>();
2490 // Resolve group columns up front so unknown names fail
2491 // before any SQL runs.
2492 let mut chosen_groups: Vec<&crate::migrate::Column> = Vec::with_capacity(group_cols.len());
2493 for name in group_cols {
2494 let col = meta
2495 .fields
2496 .iter()
2497 .find(|c| c.name == *name)
2498 .ok_or_else(|| {
2499 sqlx::Error::Protocol(format!(
2500 "umbral::orm::annotate: unknown group column `{}` on model `{}`",
2501 name,
2502 T::NAME
2503 ))
2504 })?;
2505 chosen_groups.push(col);
2506 }
2507 // Validate aggregate source columns.
2508 for (name, agg) in aggs {
2509 if let Some(col) = agg.source_column()
2510 && !meta.fields.iter().any(|c| c.name == col)
2511 {
2512 return Err(sqlx::Error::Protocol(format!(
2513 "umbral::orm::annotate: unknown column `{}` on model `{}` for aggregate `{}`",
2514 col,
2515 T::NAME,
2516 name
2517 )));
2518 }
2519 }
2520 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2521 let backend = pool.backend_name();
2522 let mut q = self.build_query_for(backend);
2523 q.clear_selects();
2524 // GROUP BY columns appear in the SELECT list AND the GROUP BY
2525 // clause. Aggregates only in the SELECT.
2526 for col in &chosen_groups {
2527 q.column(Alias::new(col.name.as_str()));
2528 q.add_group_by([sea_query::SimpleExpr::Column(sea_query::ColumnRef::Column(
2529 Alias::new(col.name.as_str()).into_iden(),
2530 ))]);
2531 }
2532 for (name, agg) in aggs {
2533 q.expr_as(agg.to_simple_expr(), Alias::new(*name));
2534 }
2535 match pool {
2536 DbPool::Sqlite(pool) => {
2537 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
2538 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, vals)
2539 .fetch_all(&pool)
2540 .await?;
2541 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2542 for row in &rows {
2543 let mut obj = serde_json::Map::with_capacity(chosen_groups.len() + aggs.len());
2544 for col in &chosen_groups {
2545 obj.insert(
2546 col.name.clone(),
2547 crate::orm::dynamic::decode_to_json(row, col)?,
2548 );
2549 }
2550 for (name, agg) in aggs {
2551 let source_ty = agg
2552 .source_column()
2553 .and_then(|c| meta.fields.iter().find(|f| f.name == c).map(|f| f.ty));
2554 obj.insert(
2555 name.to_string(),
2556 backend_sqlite::decode_agg(row, name, agg, source_ty)?,
2557 );
2558 }
2559 out.push(JsonValue::Object(obj));
2560 }
2561 Ok(out)
2562 }
2563 DbPool::Postgres(pool) => {
2564 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
2565 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, vals)
2566 .fetch_all(&pool)
2567 .await?;
2568 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2569 for row in &rows {
2570 let mut obj = serde_json::Map::with_capacity(chosen_groups.len() + aggs.len());
2571 for col in &chosen_groups {
2572 obj.insert(
2573 col.name.clone(),
2574 crate::orm::dynamic::decode_pg_to_json(row, col)?,
2575 );
2576 }
2577 for (name, agg) in aggs {
2578 let source_ty = agg
2579 .source_column()
2580 .and_then(|c| meta.fields.iter().find(|f| f.name == c).map(|f| f.ty));
2581 obj.insert(
2582 name.to_string(),
2583 backend_pg::decode_agg(row, name, agg, source_ty)?,
2584 );
2585 }
2586 out.push(JsonValue::Object(obj));
2587 }
2588 Ok(out)
2589 }
2590 }
2591 }
2592
2593 /// Django's chainable `annotate(alias=Agg("relation"))`: attach a
2594 /// related-aggregate annotation to this QuerySet. The annotation
2595 /// is **query-builder state** — it renders as a correlated scalar
2596 /// subquery inside the one SELECT every terminal builds, so it
2597 /// composes with `.filter` / `.order_by` / `.limit`, stacks with
2598 /// further annotations, and shows up in [`Self::explain`] /
2599 /// [`Self::to_sql`] out of the box. Never a side query, never an
2600 /// N+1.
2601 ///
2602 /// `relation` names a `ReverseSet` relation on the model
2603 /// (`#[umbral(reverse_fk = "...")]`), the same names
2604 /// `prefetch_related` accepts. When no declared relation matches,
2605 /// the resolver AUTO-DISCOVERS the relation (gaps2 #45): it scans
2606 /// the model registry for any child whose FK targets this parent's
2607 /// table and matches `relation` against the child's conventional
2608 /// name forms (table name, `snake_case` / lowercase struct name,
2609 /// any of those with a `_set` suffix). Declared relations always
2610 /// take precedence; an ambiguous auto-match (two children, or a
2611 /// child with two FKs to this parent) poisons the annotation with
2612 /// an error that names the candidates and points at the
2613 /// `#[umbral(reverse_fk = "...")]` escape hatch. Any
2614 /// [`crate::orm::Aggregate`] works; non-count aggregates name a
2615 /// column on the CHILD model:
2616 ///
2617 /// ```rust,ignore
2618 /// let rows = Plugin::objects()
2619 /// .filter(plugin::MODERATION.eq("approved"))
2620 /// .annotate_count("comment_set") // COUNT(*)
2621 /// .annotate_related("rating_avg", "review_set", Aggregate::avg("rating"))
2622 /// .fetch_annotated()
2623 /// .await?; // Vec<(Plugin, Map<alias, value>)>
2624 /// ```
2625 ///
2626 /// An unknown relation name doesn't panic the (infallible)
2627 /// builder — it poisons the annotation, and every fallible
2628 /// consumer (`fetch_annotated`, `explain`) reports it loudly.
2629 /// v1 caveats: child rows aggregate unconditionally — a
2630 /// child-side predicate (Django's `Count(..., filter=Q(...))`)
2631 /// and child soft-delete awareness are tracked follow-ups
2632 /// (gaps2 #39).
2633 pub fn annotate_related(
2634 mut self,
2635 alias: &str,
2636 relation: &str,
2637 agg: crate::orm::Aggregate,
2638 ) -> Self {
2639 let rev_spec = T::REVERSE_FK_RELATIONS
2640 .iter()
2641 .find(|r| r.field_name == relation);
2642 let m2m_spec = T::M2M_RELATIONS.iter().find(|r| r.field_name == relation);
2643
2644 let pk = T::FIELDS
2645 .iter()
2646 .find(|f| f.primary_key)
2647 .map(|f| f.name)
2648 .unwrap_or("id");
2649
2650 let mut child_soft_delete = false;
2651 let mut m2m_junction: Option<String> = None;
2652
2653 let resolved = if let Some(spec) = rev_spec {
2654 child_soft_delete = spec.soft_delete;
2655 Ok((
2656 spec.target_table.to_string(),
2657 spec.fk_column.to_string(),
2658 T::TABLE.to_string(),
2659 pk.to_string(),
2660 ))
2661 } else if let Some(spec) = m2m_spec {
2662 // M2M count: junction table = "<parent>_<field>", columns
2663 // parent_id / child_id. The subquery counts junction rows.
2664 m2m_junction = Some(format!("{}_{}", T::TABLE, spec.field_name));
2665 // child_table / fk_column are unused for the M2M shape, but
2666 // the tuple still carries parent_table + parent_pk for the
2667 // correlation in build_query_for.
2668 Ok((
2669 spec.target_table.to_string(),
2670 "child_id".to_string(),
2671 T::TABLE.to_string(),
2672 pk.to_string(),
2673 ))
2674 } else {
2675 // No DECLARED relation matched. Fall back to auto-discovery
2676 // (gaps2 #45): scan the model registry for any child whose
2677 // FK points back at this parent's table, and match `relation`
2678 // against the conventional name forms. Declared relations
2679 // always win above; this only runs as a fallback.
2680 match discover_reverse_relation::<T>(relation) {
2681 AutoDiscovery::Resolved {
2682 child_table,
2683 fk_column,
2684 soft_delete,
2685 } => {
2686 child_soft_delete = soft_delete;
2687 Ok((child_table, fk_column, T::TABLE.to_string(), pk.to_string()))
2688 }
2689 AutoDiscovery::Ambiguous(candidates) => Err(format!(
2690 "umbral::orm::annotate_related: ambiguous reverse relation `{relation}` on `{}` — candidates: [{}]; declare a `#[umbral(reverse_fk = \"<fk>\")] ReverseSet<Child>` field to disambiguate",
2691 T::NAME,
2692 candidates.join(", "),
2693 )),
2694 AutoDiscovery::NotFound(discoverable) => {
2695 let declared = T::REVERSE_FK_RELATIONS
2696 .iter()
2697 .map(|r| r.field_name)
2698 .collect::<Vec<_>>()
2699 .join(", ");
2700 let m2m = T::M2M_RELATIONS
2701 .iter()
2702 .map(|r| r.field_name)
2703 .collect::<Vec<_>>()
2704 .join(", ");
2705 Err(format!(
2706 "umbral::orm::annotate_related: `{relation}` is not a reverse-FK or M2M relation on `{}` — reverse-FK relations: [{declared}], M2M relations: [{m2m}], auto-discoverable children: [{}]",
2707 T::NAME,
2708 discoverable.join(", "),
2709 ))
2710 }
2711 }
2712 };
2713
2714 self.annotations.push(RelatedAnnotation {
2715 alias: alias.to_string(),
2716 agg,
2717 resolved,
2718 child_soft_delete,
2719 child_filter: None,
2720 m2m_junction,
2721 });
2722 self
2723 }
2724
2725 /// Sugar for the overwhelmingly common annotation:
2726 /// `annotate_related("<relation>_count", relation, Aggregate::count())`.
2727 /// `.annotate_count("comment_set")` exposes the value under the
2728 /// `comment_set_count` alias in [`Self::fetch_annotated`].
2729 pub fn annotate_count(self, relation: &str) -> Self {
2730 let alias = format!("{relation}_count");
2731 self.annotate_related(&alias, relation, crate::orm::Aggregate::count())
2732 }
2733
2734 /// Like [`Self::annotate_count`] but counts only the children
2735 /// matching `pred` — Django's `Count("comments", filter=Q(...))`.
2736 /// `C` is the CHILD model, so the predicate is typed against the
2737 /// child's columns (`comment::MODERATION.eq("visible")`). The
2738 /// predicate renders into the correlated count subquery's WHERE
2739 /// alongside the FK correlation and the auto soft-delete filter.
2740 ///
2741 /// ```rust,ignore
2742 /// Plugin::objects()
2743 /// .annotate_count_where::<PluginComment>(
2744 /// "visible_comments",
2745 /// "comment_set",
2746 /// plugin_comment::MODERATION.eq("visible"),
2747 /// )
2748 /// ```
2749 pub fn annotate_count_where<C: crate::orm::Model>(
2750 self,
2751 alias: &str,
2752 relation: &str,
2753 pred: crate::orm::Predicate<C>,
2754 ) -> Self {
2755 // Render the child predicate to a backend-default SimpleExpr.
2756 // The count subquery embeds one expression; the equality /
2757 // comparison predicates used for child filters render the same
2758 // on both backends, so the default `cond` is correct.
2759 let child_filter = pred.cond_for("postgres");
2760 let mut queryset = self.annotate_related(alias, relation, crate::orm::Aggregate::count());
2761 // The just-pushed annotation is the last one; attach the filter.
2762 if let Some(last) = queryset.annotations.last_mut() {
2763 last.child_filter = Some(child_filter);
2764 }
2765 queryset
2766 }
2767
2768 /// Loud-failure check for poisoned annotations (unknown relation
2769 /// names recorded by the infallible builder). Called by every
2770 /// fallible consumer before SQL runs.
2771 fn check_annotations(&self) -> Result<(), sqlx::Error> {
2772 for ann in &self.annotations {
2773 if let Err(msg) = &ann.resolved {
2774 return Err(sqlx::Error::Protocol(msg.clone()));
2775 }
2776 }
2777 Ok(())
2778 }
2779
2780 /// Run the SELECT and return every matching row **with its
2781 /// annotation values** — the execution terminal for
2782 /// [`Self::annotate_related`] / [`Self::annotate_count`]. One
2783 /// query; each row's annotations arrive as an `alias → JSON
2784 /// value` map (count → integer, AVG → float/null, SUM/MAX/MIN →
2785 /// typed per the child column, NULL on empty sets for the
2786 /// non-count aggregates).
2787 pub async fn fetch_annotated(
2788 self,
2789 ) -> Result<Vec<(T, serde_json::Map<String, JsonValue>)>, sqlx::Error>
2790 where
2791 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
2792 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
2793 {
2794 self.check_annotations()?;
2795 // Child-column types for SUM/MAX/MIN decoding, resolved from
2796 // the runtime registry (the child model is known by table
2797 // name only). `None` falls back to decode_agg's string path.
2798 let source_types: Vec<(String, crate::orm::Aggregate, Option<crate::orm::SqlType>)> = {
2799 let registry_up = crate::migrate::is_initialised();
2800 self.annotations
2801 .iter()
2802 .map(|ann| {
2803 let ty = match (&ann.resolved, registry_up, ann.agg.source_column()) {
2804 (Ok((child_table, ..)), true, Some(col)) => {
2805 crate::migrate::registered_models()
2806 .into_iter()
2807 .find(|m| m.table == *child_table)
2808 .and_then(|m| m.fields.iter().find(|f| f.name == col).map(|f| f.ty))
2809 }
2810 _ => None,
2811 };
2812 (ann.alias.clone(), ann.agg.clone(), ty)
2813 })
2814 .collect()
2815 };
2816
2817 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2818 match pool {
2819 DbPool::Sqlite(pool) => {
2820 let q = self.build_query_for("sqlite");
2821 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
2822 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, vals)
2823 .fetch_all(&pool)
2824 .await?;
2825 let mut out = Vec::with_capacity(rows.len());
2826 for row in &rows {
2827 let t = <T as sqlx::FromRow<_>>::from_row(row)?;
2828 let mut anns = serde_json::Map::with_capacity(source_types.len());
2829 for (alias, agg, ty) in &source_types {
2830 anns.insert(
2831 alias.clone(),
2832 backend_sqlite::decode_agg(row, alias, agg, *ty)?,
2833 );
2834 }
2835 out.push((t, anns));
2836 }
2837 Ok(out)
2838 }
2839 DbPool::Postgres(pool) => {
2840 let q = self.build_query_for("postgres");
2841 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
2842 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, vals)
2843 .fetch_all(&pool)
2844 .await?;
2845 let mut out = Vec::with_capacity(rows.len());
2846 for row in &rows {
2847 let t = <T as sqlx::FromRow<_>>::from_row(row)?;
2848 let mut anns = serde_json::Map::with_capacity(source_types.len());
2849 for (alias, agg, ty) in &source_types {
2850 anns.insert(alias.clone(), backend_pg::decode_agg(row, alias, agg, *ty)?);
2851 }
2852 out.push((t, anns));
2853 }
2854 Ok(out)
2855 }
2856 }
2857 }
2858
2859 /// `DELETE FROM table WHERE <predicates>`. Returns the number of
2860 /// rows deleted. With no `.filter` calls, deletes every row.
2861 ///
2862 /// Fires `bulk_post_delete:<table>` once with the list of removed
2863 /// PKs when at least one row was deleted. Per-row `pre_delete` /
2864 /// `post_delete` are NOT fired by this path — use
2865 /// [`Manager::delete_instance`] when per-row signal semantics are
2866 /// required.
2867 ///
2868 /// Feature #72: for `#[umbral(soft_delete)]` models this rewrites
2869 /// to `UPDATE ... SET deleted_at = NOW() WHERE ...` so rows
2870 /// survive in the DB (filtered out of subsequent queries by the
2871 /// auto `WHERE deleted_at IS NULL`). Call `.hard_delete()`
2872 /// beforehand for a real DELETE (GDPR purge, test cleanup).
2873 pub async fn delete(self) -> Result<u64, sqlx::Error> {
2874 // Feature #72 — soft-delete redirect. The whole `delete()`
2875 // contract collapses to an UPDATE setting `deleted_at`. We
2876 // keep the bulk_post_delete signal so subscribers see the
2877 // same event shape regardless of the underlying SQL.
2878 if self.soft_delete_active && !self.hard_delete {
2879 return self.soft_delete_update().await;
2880 }
2881 let atomic = self.should_atomic_wrap();
2882 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Write);
2883 let backend = pool.backend_name();
2884 let mut stmt = self.build_delete_for(backend);
2885 let pk = pk_field::<T>();
2886 if let Some(field) = pk {
2887 stmt.returning_col(Alias::new(field.name));
2888 }
2889 let ids: Vec<JsonValue> = match pool {
2890 DbPool::Sqlite(pool) => {
2891 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
2892 let rows = if atomic {
2893 let mut tx = pool.begin().await?;
2894 let r = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
2895 .fetch_all(&mut *tx)
2896 .await;
2897 match r {
2898 Ok(rows) => {
2899 tx.commit().await?;
2900 rows
2901 }
2902 Err(e) => {
2903 let _ = tx.rollback().await;
2904 return Err(e);
2905 }
2906 }
2907 } else {
2908 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
2909 .fetch_all(&pool)
2910 .await?
2911 };
2912 match pk {
2913 Some(field) => rows
2914 .iter()
2915 .map(|r| backend_sqlite::pk_to_json(r, field.name, field.ty))
2916 .collect::<Result<_, _>>()?,
2917 None => Vec::new(),
2918 }
2919 }
2920 DbPool::Postgres(pool) => {
2921 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
2922 let rows = if atomic {
2923 let mut tx = pool.begin().await?;
2924 let r = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
2925 .fetch_all(&mut *tx)
2926 .await;
2927 match r {
2928 Ok(rows) => {
2929 tx.commit().await?;
2930 rows
2931 }
2932 Err(e) => {
2933 let _ = tx.rollback().await;
2934 return Err(e);
2935 }
2936 }
2937 } else {
2938 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
2939 .fetch_all(&pool)
2940 .await?
2941 };
2942 match pk {
2943 Some(field) => rows
2944 .iter()
2945 .map(|r| backend_pg::pk_to_json(r, field.name, field.ty))
2946 .collect::<Result<_, _>>()?,
2947 None => Vec::new(),
2948 }
2949 }
2950 };
2951 let count = ids.len() as u64;
2952 if !ids.is_empty() {
2953 crate::signals::emit_bulk_post_delete::<T>(ids).await;
2954 }
2955 Ok(count)
2956 }
2957
2958 /// `UPDATE table SET col = <expr> WHERE <predicates>` using an
2959 /// F-expression for the new value.
2960 ///
2961 /// This is the companion to `update_values` for atomic column
2962 /// arithmetic. `F::col("views").add(1)` produces an [`FExpr`] that
2963 /// renders as `SET views = views + 1` — the database computes the
2964 /// increment atomically on the server side rather than needing a
2965 /// read-modify-write round-trip in application code.
2966 ///
2967 /// ```rust,ignore
2968 /// use umbral::orm::F;
2969 ///
2970 /// Post::objects()
2971 /// .filter(post::ID.eq(42))
2972 /// .update_expr("views", F::col("views").add(1))
2973 /// .await?;
2974 /// ```
2975 ///
2976 /// Mixing `update_values` and `update_expr` for different columns in
2977 /// one statement requires two separate calls. A combined API (a map
2978 /// where values can be either JSON or FExpr) would require a new sum
2979 /// type; deferred until a consumer surfaces the need.
2980 pub async fn update_expr(
2981 self,
2982 col_name: &str,
2983 expr: FExpr,
2984 ) -> Result<u64, crate::orm::write::WriteError> {
2985 use crate::orm::write::WriteError;
2986 // Validate the column exists on the model.
2987 let field = T::FIELDS
2988 .iter()
2989 .find(|f| f.name == col_name)
2990 .ok_or_else(|| WriteError::UnknownColumn {
2991 field: col_name.to_string(),
2992 })?;
2993 if field.primary_key {
2994 // Silently skip PK rewrites, same as update_values.
2995 return Ok(0);
2996 }
2997 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Write);
2998 let backend = pool.backend_name();
2999
3000 let mut stmt = sea_query::Query::update();
3001 stmt.table(crate::db::router::schema_qualified_table(T::TABLE));
3002 stmt.value(Alias::new(field.name), expr.to_simple_expr());
3003 for p in &self.predicates {
3004 stmt.and_where(p.cond_for(backend));
3005 }
3006 if self.soft_delete_active {
3007 if self.only_deleted {
3008 stmt.and_where(Expr::col(Alias::new("deleted_at")).is_not_null());
3009 } else if !self.with_deleted {
3010 stmt.and_where(Expr::col(Alias::new("deleted_at")).is_null());
3011 }
3012 }
3013 let pk = pk_field::<T>();
3014 if let Some(pkf) = pk {
3015 stmt.returning_col(Alias::new(pkf.name));
3016 }
3017
3018 let ids: Vec<JsonValue> = match pool {
3019 DbPool::Sqlite(pool) => {
3020 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
3021 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3022 .fetch_all(&pool)
3023 .await?;
3024 match pk {
3025 Some(pkf) => rows
3026 .iter()
3027 .map(|r| backend_sqlite::pk_to_json(r, pkf.name, pkf.ty))
3028 .collect::<Result<_, _>>()
3029 .map_err(crate::orm::write::WriteError::Sqlx)?,
3030 None => Vec::new(),
3031 }
3032 }
3033 DbPool::Postgres(pool) => {
3034 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
3035 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3036 .fetch_all(&pool)
3037 .await?;
3038 match pk {
3039 Some(pkf) => rows
3040 .iter()
3041 .map(|r| backend_pg::pk_to_json(r, pkf.name, pkf.ty))
3042 .collect::<Result<_, _>>()
3043 .map_err(crate::orm::write::WriteError::Sqlx)?,
3044 None => Vec::new(),
3045 }
3046 }
3047 };
3048 let count = ids.len() as u64;
3049 if !ids.is_empty() {
3050 crate::signals::emit_bulk_post_save::<T>(ids, false).await;
3051 }
3052 Ok(count)
3053 }
3054
3055 /// `UPDATE table SET k=v[, ...] WHERE <predicates>`. The values
3056 /// map provides `column_name → JSON value` pairs; each is
3057 /// converted to a `sea_query::Value` per the column's declared
3058 /// `SqlType` via [`crate::orm::write::json_to_sea_value`]. Returns
3059 /// the number of rows affected.
3060 ///
3061 /// Unknown columns in the map fail loudly with
3062 /// `WriteError::UnknownColumn`. JSON `null` is rejected for
3063 /// non-nullable columns; supplying a column that exists but is
3064 /// absent from the map is silently a no-op (the column keeps its
3065 /// current value — PATCH semantics, not PUT).
3066 ///
3067 /// Fires `bulk_post_save:<table>` once with `{ ids, created:
3068 /// false, actor }` when at least one row matched. Per-row
3069 /// `pre_save` / `post_save` are NOT fired — use [`Manager::save`]
3070 /// when per-row signal semantics are required.
3071 pub async fn update_values(
3072 self,
3073 values: serde_json::Map<String, serde_json::Value>,
3074 ) -> Result<u64, crate::orm::write::WriteError> {
3075 let atomic = self.should_atomic_wrap();
3076 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Write);
3077 let backend = pool.backend_name();
3078 let mut stmt = self.build_update_for(backend, &values)?;
3079 // RETURNING <pk> so bulk_post_save can include the matched ids.
3080 let pk = pk_field::<T>();
3081 if let Some(field) = pk {
3082 stmt.returning_col(Alias::new(field.name));
3083 }
3084 let ids: Vec<JsonValue> = match pool {
3085 DbPool::Sqlite(pool) => {
3086 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
3087 let rows = if atomic {
3088 let mut tx = pool
3089 .begin()
3090 .await
3091 .map_err(crate::orm::write::WriteError::Sqlx)?;
3092 let r = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3093 .fetch_all(&mut *tx)
3094 .await;
3095 match r {
3096 Ok(rows) => {
3097 tx.commit()
3098 .await
3099 .map_err(crate::orm::write::WriteError::Sqlx)?;
3100 rows
3101 }
3102 Err(e) => {
3103 let _ = tx.rollback().await;
3104 return Err(crate::orm::write::WriteError::Sqlx(e));
3105 }
3106 }
3107 } else {
3108 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3109 .fetch_all(&pool)
3110 .await?
3111 };
3112 match pk {
3113 Some(field) => rows
3114 .iter()
3115 .map(|r| backend_sqlite::pk_to_json(r, field.name, field.ty))
3116 .collect::<Result<_, _>>()
3117 .map_err(crate::orm::write::WriteError::Sqlx)?,
3118 None => Vec::new(),
3119 }
3120 }
3121 DbPool::Postgres(pool) => {
3122 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
3123 let rows = if atomic {
3124 let mut tx = pool
3125 .begin()
3126 .await
3127 .map_err(crate::orm::write::WriteError::Sqlx)?;
3128 let r = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3129 .fetch_all(&mut *tx)
3130 .await;
3131 match r {
3132 Ok(rows) => {
3133 tx.commit()
3134 .await
3135 .map_err(crate::orm::write::WriteError::Sqlx)?;
3136 rows
3137 }
3138 Err(e) => {
3139 let _ = tx.rollback().await;
3140 return Err(crate::orm::write::WriteError::Sqlx(e));
3141 }
3142 }
3143 } else {
3144 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3145 .fetch_all(&pool)
3146 .await?
3147 };
3148 match pk {
3149 Some(field) => rows
3150 .iter()
3151 .map(|r| backend_pg::pk_to_json(r, field.name, field.ty))
3152 .collect::<Result<_, _>>()
3153 .map_err(crate::orm::write::WriteError::Sqlx)?,
3154 None => Vec::new(),
3155 }
3156 }
3157 };
3158 let count = ids.len() as u64;
3159 if !ids.is_empty() {
3160 crate::signals::emit_bulk_post_save::<T>(ids, false).await;
3161 }
3162 Ok(count)
3163 }
3164
3165 /// Helper: build the DELETE statement for the active backend.
3166 /// Public-by-virtue-of-being-pub(crate) so the `_pg` and (future)
3167 /// `_sqlite` explicit-pool variants can share the SQL builder.
3168 fn build_delete_for(&self, backend_name: &str) -> sea_query::DeleteStatement {
3169 let mut stmt = Query::delete();
3170 stmt.from_table(crate::db::router::schema_qualified_table(T::TABLE));
3171 for p in &self.predicates {
3172 stmt.and_where(p.cond_for(backend_name));
3173 }
3174 stmt
3175 }
3176
3177 /// Feature #72 — soft-delete rewrite: turn `DELETE FROM table
3178 /// WHERE ...` into `UPDATE table SET deleted_at = NOW() WHERE
3179 /// ... AND deleted_at IS NULL`. The trailing `IS NULL` guard
3180 /// makes the operation idempotent: re-soft-deleting an already-
3181 /// soft-deleted row doesn't bump its timestamp. Fires
3182 /// `bulk_post_delete:<table>` so subscribers see the same event
3183 /// shape as a hard delete.
3184 async fn soft_delete_update(self) -> Result<u64, sqlx::Error> {
3185 let atomic = self.should_atomic_wrap();
3186 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Write);
3187 let backend = pool.backend_name();
3188 let now = chrono::Utc::now();
3189 let mut stmt = sea_query::Query::update();
3190 stmt.table(crate::db::router::schema_qualified_table(T::TABLE));
3191 stmt.value(
3192 Alias::new("deleted_at"),
3193 sea_query::Value::ChronoDateTimeUtc(Some(Box::new(now))),
3194 );
3195 for p in &self.predicates {
3196 stmt.and_where(p.cond_for(backend));
3197 }
3198 // Idempotency guard — never bump an already-set deleted_at.
3199 stmt.and_where(sea_query::Expr::col(Alias::new("deleted_at")).is_null());
3200 let pk = pk_field::<T>();
3201 if let Some(pkf) = pk {
3202 stmt.returning_col(Alias::new(pkf.name));
3203 }
3204 let ids: Vec<JsonValue> = match pool {
3205 DbPool::Sqlite(pool) => {
3206 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
3207 let rows = if atomic {
3208 let mut tx = pool.begin().await?;
3209 let r = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3210 .fetch_all(&mut *tx)
3211 .await;
3212 match r {
3213 Ok(rows) => {
3214 tx.commit().await?;
3215 rows
3216 }
3217 Err(e) => {
3218 let _ = tx.rollback().await;
3219 return Err(e);
3220 }
3221 }
3222 } else {
3223 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3224 .fetch_all(&pool)
3225 .await?
3226 };
3227 match pk {
3228 Some(field) => rows
3229 .iter()
3230 .map(|r| backend_sqlite::pk_to_json(r, field.name, field.ty))
3231 .collect::<Result<_, _>>()?,
3232 None => Vec::new(),
3233 }
3234 }
3235 DbPool::Postgres(pool) => {
3236 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
3237 let rows = if atomic {
3238 let mut tx = pool.begin().await?;
3239 let r = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3240 .fetch_all(&mut *tx)
3241 .await;
3242 match r {
3243 Ok(rows) => {
3244 tx.commit().await?;
3245 rows
3246 }
3247 Err(e) => {
3248 let _ = tx.rollback().await;
3249 return Err(e);
3250 }
3251 }
3252 } else {
3253 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3254 .fetch_all(&pool)
3255 .await?
3256 };
3257 match pk {
3258 Some(field) => rows
3259 .iter()
3260 .map(|r| backend_pg::pk_to_json(r, field.name, field.ty))
3261 .collect::<Result<_, _>>()?,
3262 None => Vec::new(),
3263 }
3264 }
3265 };
3266 let count = ids.len() as u64;
3267 if !ids.is_empty() {
3268 crate::signals::emit_bulk_post_delete::<T>(ids).await;
3269 }
3270 Ok(count)
3271 }
3272
3273 /// Helper: build the UPDATE statement for the active backend.
3274 /// Walks the `values` map, validates each column against the
3275 /// model's `FIELDS` metadata, converts the JSON value via
3276 /// `write::json_to_sea_value`, and threads the accumulated
3277 /// predicates into the WHERE clause.
3278 fn build_update_for(
3279 &self,
3280 backend_name: &str,
3281 values: &serde_json::Map<String, serde_json::Value>,
3282 ) -> Result<sea_query::UpdateStatement, crate::orm::write::WriteError> {
3283 use crate::orm::write::{WriteError, json_to_sea_value};
3284 let mut stmt = Query::update();
3285 stmt.table(crate::db::router::schema_qualified_table(T::TABLE));
3286 for (col_name, val) in values {
3287 // Look up the column on the model. Unknown column names
3288 // fail loudly here rather than producing a bad UPDATE.
3289 let field = T::FIELDS
3290 .iter()
3291 .find(|f| f.name == col_name.as_str())
3292 .ok_or_else(|| WriteError::UnknownColumn {
3293 field: col_name.clone(),
3294 })?;
3295 // Reject attempts to overwrite the PK via update_values.
3296 // The QuerySet's WHERE clause is the only way to identify
3297 // rows; rewriting the PK while filtering on the old one
3298 // is a footgun.
3299 if field.primary_key {
3300 continue;
3301 }
3302 let sea_value =
3303 json_to_sea_value(field.ty, val, field.nullable, field.name, fk_pk_hint(field))?;
3304 stmt.value(Alias::new(field.name), sea_value);
3305 }
3306 for p in &self.predicates {
3307 stmt.and_where(p.cond_for(backend_name));
3308 }
3309 if self.soft_delete_active {
3310 if self.only_deleted {
3311 stmt.and_where(Expr::col(Alias::new("deleted_at")).is_not_null());
3312 } else if !self.with_deleted {
3313 stmt.and_where(Expr::col(Alias::new("deleted_at")).is_null());
3314 }
3315 }
3316 Ok(stmt)
3317 }
3318
3319 /// Run the SELECT against an explicit `PgPool` and return every
3320 /// matching row. Bound by `FromRow<PgRow>` alone so models with
3321 /// Postgres-only field types compile.
3322 pub async fn fetch_pg(self, pool: &sqlx::PgPool) -> Result<Vec<T>, sqlx::Error>
3323 where
3324 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3325 {
3326 let q = self.build_query_for("postgres");
3327 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
3328 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
3329 .fetch_all(pool)
3330 .await
3331 }
3332
3333 /// Run the SELECT against an explicit `PgPool` with LIMIT 1.
3334 pub async fn first_pg(mut self, pool: &sqlx::PgPool) -> Result<Option<T>, sqlx::Error>
3335 where
3336 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3337 {
3338 self.query.limit(1);
3339 let q = self.build_query_for("postgres");
3340 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
3341 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
3342 .fetch_optional(pool)
3343 .await
3344 }
3345
3346 /// Run `SELECT COUNT(*)` against an explicit `PgPool`. No FromRow
3347 /// bound on `T` — the count tuple type is `(i64,)`.
3348 pub async fn count_pg(self, pool: &sqlx::PgPool) -> Result<i64, sqlx::Error> {
3349 let mut rebuilt = self.build_query_for("postgres");
3350 rebuilt.clear_selects();
3351 rebuilt.expr(Func::count(Expr::col(sea_query::Asterisk)));
3352 rebuilt.reset_limit();
3353 rebuilt.reset_offset();
3354 let (sql, values) = rebuilt.build_sqlx(PostgresQueryBuilder);
3355 let (n,): (i64,) = sqlx::query_as_with::<sqlx::Postgres, (i64,), _>(&sql, values)
3356 .fetch_one(pool)
3357 .await?;
3358 Ok(n)
3359 }
3360
3361 /// Return whether any row matches, against an explicit `PgPool`.
3362 pub async fn exists_pg(self, pool: &sqlx::PgPool) -> Result<bool, sqlx::Error>
3363 where
3364 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3365 {
3366 let rows = self.limit(1).fetch_pg(pool).await?;
3367 Ok(!rows.is_empty())
3368 }
3369
3370 /// Exactly-one terminal against an explicit `PgPool`.
3371 /// See [`QuerySet::get`] for the error-variant semantics.
3372 pub async fn get_pg(self, pool: &sqlx::PgPool) -> Result<T, GetError>
3373 where
3374 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3375 {
3376 let mut rows = self.limit(2).fetch_pg(pool).await.map_err(GetError::Sqlx)?;
3377 match rows.len() {
3378 0 => Err(GetError::NotFound),
3379 1 => Ok(rows.pop().unwrap()),
3380 _ => Err(GetError::MultipleObjectsReturned),
3381 }
3382 }
3383}
3384
3385/// Delegating chainable + terminal surface on `Manager<T>`.
3386///
3387/// Lets users write `Post::objects().filter(...).fetch().await` without
3388/// a separate `.query()` hop. Each method constructs the initial
3389/// `SelectStatement` against `T::TABLE` with one column per
3390/// `T::FIELDS` entry, wraps it in a fresh `QuerySet<T>`, and forwards.
3391impl<T: Model> Manager<T> {
3392 fn queryset(&self) -> QuerySet<T> {
3393 let columns: Vec<Alias> = T::FIELDS.iter().map(|f| Alias::new(f.name)).collect();
3394 let query = Query::select()
3395 .columns(columns)
3396 .from(crate::db::router::schema_qualified_table(T::TABLE))
3397 .take();
3398 let mut qs = QuerySet::new(query);
3399 // BUG-8: seed the default ORDER BY from `Model::ORDERING` so
3400 // terminals that don't see an explicit `.order_by(...)` still
3401 // get a deterministic row order.
3402 qs.default_ordering = T::ORDERING.to_vec();
3403 // Propagate the Manager's atomic override so QuerySet
3404 // terminals inherit it without the caller re-specifying.
3405 qs.atomic = self.atomic;
3406 // Feature #72 — snapshot the model's soft-delete opt-in into
3407 // the QuerySet so the build_query_for path knows whether to
3408 // auto-inject `WHERE deleted_at IS NULL`. Without this
3409 // snapshot the `impl<T> QuerySet<T>` path can't read
3410 // `T::SOFT_DELETE` (T is unbounded there).
3411 qs.soft_delete_active = T::SOFT_DELETE;
3412 qs
3413 }
3414
3415 /// Resolve whether write terminals on this Manager should auto-wrap
3416 /// in a transaction. Per-call override > builder global.
3417 fn should_atomic_wrap(&self) -> bool {
3418 self.atomic.unwrap_or_else(crate::db::atomic_default)
3419 }
3420
3421 /// See `QuerySet::filter`.
3422 pub fn filter(&self, p: Predicate<T>) -> QuerySet<T> {
3423 self.queryset().filter(p)
3424 }
3425
3426 /// See `QuerySet::exclude`.
3427 pub fn exclude(&self, p: Predicate<T>) -> QuerySet<T> {
3428 self.queryset().exclude(p)
3429 }
3430
3431 /// A bare [`QuerySet`] over every row — Django's `Model.objects.all()`.
3432 ///
3433 /// The entry point when you need a `QuerySet` terminal without a
3434 /// filter: a grouped aggregate over the whole table, an unfiltered
3435 /// `aggregate`, or just an explicit "all rows" for readability.
3436 pub fn all(&self) -> QuerySet<T> {
3437 self.queryset()
3438 }
3439
3440 /// See [`QuerySet::aggregate`] — single-row aggregate over every row.
3441 ///
3442 /// Forwards from the manager so `Model::objects().aggregate(...)`
3443 /// works without an intervening `.filter(...)` / `.on(...)`.
3444 pub async fn aggregate(
3445 &self,
3446 aggs: &[(&str, crate::orm::Aggregate)],
3447 ) -> Result<JsonValue, sqlx::Error> {
3448 self.queryset().aggregate(aggs).await
3449 }
3450
3451 /// See [`QuerySet::annotate`] — grouped aggregate (`GROUP BY <group_cols>`).
3452 ///
3453 /// Forwards from the manager so the documented
3454 /// `Model::objects().annotate(&["status"], &[("count", Aggregate::count())])`
3455 /// (Django's `.values("status").annotate(count=Count("id"))`) compiles
3456 /// directly, without a filter first.
3457 pub async fn annotate(
3458 &self,
3459 group_cols: &[&str],
3460 aggs: &[(&str, crate::orm::Aggregate)],
3461 ) -> Result<Vec<JsonValue>, sqlx::Error> {
3462 self.queryset().annotate(group_cols, aggs).await
3463 }
3464
3465 /// Feature #72 — see `QuerySet::with_deleted`.
3466 pub fn with_deleted(&self) -> QuerySet<T> {
3467 self.queryset().with_deleted()
3468 }
3469
3470 /// Feature #72 — see `QuerySet::only_deleted`.
3471 pub fn only_deleted(&self) -> QuerySet<T> {
3472 self.queryset().only_deleted()
3473 }
3474
3475 /// Gap #111 — see [`QuerySet::only`].
3476 pub fn only(&self, cols: &[&str]) -> QuerySet<T> {
3477 self.queryset().only(cols)
3478 }
3479
3480 /// See [`QuerySet::join_related`].
3481 pub fn join_related(&self, field_name: impl Into<String>) -> QuerySet<T> {
3482 self.queryset().join_related(field_name)
3483 }
3484
3485 /// See [`QuerySet::join_related_many`].
3486 pub fn join_related_many(&self, field_names: &[&str]) -> QuerySet<T> {
3487 self.queryset().join_related_many(field_names)
3488 }
3489
3490 /// See [`QuerySet::left_join_related`].
3491 pub fn left_join_related(&self, path: impl Into<String>) -> QuerySet<T> {
3492 self.queryset().left_join_related(path)
3493 }
3494
3495 /// See [`QuerySet::inner_join_related`].
3496 pub fn inner_join_related(&self, path: impl Into<String>) -> QuerySet<T> {
3497 self.queryset().inner_join_related(path)
3498 }
3499
3500 /// See [`QuerySet::right_join_related`].
3501 pub fn right_join_related(&self, path: impl Into<String>) -> QuerySet<T> {
3502 self.queryset().right_join_related(path)
3503 }
3504
3505 /// See [`QuerySet::select_related`].
3506 pub fn select_related(&self, field_name: impl Into<String>) -> QuerySet<T> {
3507 self.queryset().select_related(field_name)
3508 }
3509
3510 /// See [`QuerySet::select_related_many`].
3511 pub fn select_related_many(&self, field_names: &[&str]) -> QuerySet<T> {
3512 self.queryset().select_related_many(field_names)
3513 }
3514
3515 /// See [`QuerySet::prefetch_related`].
3516 pub fn prefetch_related(&self, field_name: impl Into<String>) -> QuerySet<T> {
3517 self.queryset().prefetch_related(field_name)
3518 }
3519
3520 /// See [`QuerySet::prefetch_related_many`].
3521 pub fn prefetch_related_many(&self, field_names: &[&str]) -> QuerySet<T> {
3522 self.queryset().prefetch_related_many(field_names)
3523 }
3524
3525 /// Feature #72 — see `QuerySet::hard_delete`.
3526 pub fn hard_delete(&self) -> QuerySet<T> {
3527 self.queryset().hard_delete()
3528 }
3529
3530 /// See `QuerySet::into_subquery`.
3531 pub fn into_subquery(&self, col_name: &str) -> crate::orm::Subquery {
3532 self.queryset().into_subquery(col_name)
3533 }
3534
3535 /// See `QuerySet::order_by`.
3536 pub fn order_by(&self, o: OrderExpr<T>) -> QuerySet<T> {
3537 self.queryset().order_by(o)
3538 }
3539
3540 /// See `QuerySet::limit`.
3541 pub fn limit(&self, n: u64) -> QuerySet<T> {
3542 self.queryset().limit(n)
3543 }
3544
3545 /// See `QuerySet::offset`.
3546 pub fn offset(&self, n: u64) -> QuerySet<T> {
3547 self.queryset().offset(n)
3548 }
3549
3550 /// See `QuerySet::on`.
3551 pub fn on(&self, pool: &sqlx::SqlitePool) -> QuerySet<T> {
3552 self.queryset().on(pool)
3553 }
3554
3555 /// See `QuerySet::on_pg`.
3556 pub fn on_pg(&self, pool: &sqlx::PgPool) -> QuerySet<T> {
3557 self.queryset().on_pg(pool)
3558 }
3559
3560 /// See `QuerySet::fetch`.
3561 pub async fn fetch(&self) -> Result<Vec<T>, sqlx::Error>
3562 where
3563 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3564 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
3565 + HydrateRelated,
3566 {
3567 self.queryset().fetch().await
3568 }
3569
3570 /// See `QuerySet::first`.
3571 pub async fn first(&self) -> Result<Option<T>, sqlx::Error>
3572 where
3573 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3574 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
3575 + HydrateRelated,
3576 {
3577 self.queryset().first().await
3578 }
3579
3580 /// See `QuerySet::count`.
3581 pub async fn count(&self) -> Result<i64, sqlx::Error> {
3582 self.queryset().count().await
3583 }
3584
3585 /// See [`QuerySet::annotate_related`] — starts an annotated chain
3586 /// from the manager, like `filter` does.
3587 pub fn annotate_related(
3588 &self,
3589 alias: &str,
3590 relation: &str,
3591 agg: crate::orm::Aggregate,
3592 ) -> QuerySet<T> {
3593 self.queryset().annotate_related(alias, relation, agg)
3594 }
3595
3596 /// See [`QuerySet::annotate_count`].
3597 pub fn annotate_count(&self, relation: &str) -> QuerySet<T> {
3598 self.queryset().annotate_count(relation)
3599 }
3600
3601 /// See [`QuerySet::annotate_count_where`] — starts a filtered
3602 /// annotated chain from the manager.
3603 pub fn annotate_count_where<C: crate::orm::Model>(
3604 &self,
3605 alias: &str,
3606 relation: &str,
3607 pred: crate::orm::Predicate<C>,
3608 ) -> QuerySet<T> {
3609 self.queryset()
3610 .annotate_count_where::<C>(alias, relation, pred)
3611 }
3612
3613 /// See [`QuerySet::fetch_annotated`].
3614 pub async fn fetch_annotated(
3615 &self,
3616 ) -> Result<Vec<(T, serde_json::Map<String, JsonValue>)>, sqlx::Error>
3617 where
3618 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3619 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3620 {
3621 self.queryset().fetch_annotated().await
3622 }
3623
3624 /// See [`QuerySet::values`].
3625 pub async fn values(&self, columns: &[&str]) -> Result<Vec<JsonValue>, sqlx::Error> {
3626 self.queryset().values(columns).await
3627 }
3628
3629 /// See `QuerySet::exists`.
3630 pub async fn exists(&self) -> Result<bool, sqlx::Error>
3631 where
3632 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3633 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
3634 + HydrateRelated,
3635 {
3636 self.queryset().exists().await
3637 }
3638
3639 /// `.get(predicate)` — sugar for `.filter(predicate).get()`.
3640 ///
3641 /// The Django-shape one-liner: `User::objects().get(user::ID.eq(1))`.
3642 /// See [`QuerySet::get`] for error-variant semantics.
3643 pub async fn get(&self, p: Predicate<T>) -> Result<T, GetError>
3644 where
3645 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3646 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
3647 + HydrateRelated,
3648 {
3649 self.queryset().filter(p).get().await
3650 }
3651
3652 /// See [`QuerySet::fetch_pg`].
3653 pub async fn fetch_pg(&self, pool: &sqlx::PgPool) -> Result<Vec<T>, sqlx::Error>
3654 where
3655 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3656 {
3657 self.queryset().fetch_pg(pool).await
3658 }
3659
3660 /// See [`QuerySet::first_pg`].
3661 pub async fn first_pg(&self, pool: &sqlx::PgPool) -> Result<Option<T>, sqlx::Error>
3662 where
3663 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3664 {
3665 self.queryset().first_pg(pool).await
3666 }
3667
3668 /// See [`QuerySet::count_pg`].
3669 pub async fn count_pg(&self, pool: &sqlx::PgPool) -> Result<i64, sqlx::Error> {
3670 self.queryset().count_pg(pool).await
3671 }
3672
3673 /// See [`QuerySet::exists_pg`].
3674 pub async fn exists_pg(&self, pool: &sqlx::PgPool) -> Result<bool, sqlx::Error>
3675 where
3676 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3677 {
3678 self.queryset().exists_pg(pool).await
3679 }
3680
3681 /// Postgres-only sugar for `.filter(predicate).get_pg(pool)`.
3682 pub async fn get_pg(&self, pool: &sqlx::PgPool, p: Predicate<T>) -> Result<T, GetError>
3683 where
3684 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3685 {
3686 self.queryset().filter(p).get_pg(pool).await
3687 }
3688
3689 // =====================================================================
3690 // Write methods — INSERT.
3691 //
3692 // `create(instance)` does one row; `bulk_create([...])` does many in
3693 // a single multi-VALUES INSERT. Both serialise the instance(s) to a
3694 // JSON map via `serde::Serialize`, look up each field in the model's
3695 // `FIELDS` metadata, and bind values through
3696 // [`crate::orm::write::json_to_sea_value`].
3697 //
3698 // PK handling:
3699 // - Default value (0 for ints, nil for UUIDs, empty for String):
3700 // omitted from the INSERT column list so the DB autoincrement /
3701 // default kicks in.
3702 // - Explicit non-default value: included in the INSERT so the
3703 // caller can supply UUIDs / slug PKs themselves.
3704 // =====================================================================
3705
3706 /// INSERT one row, return the row as it now exists in the
3707 /// database (with any autoincrement PK populated). Uses the
3708 /// ambient pool via `Manager::queryset().resolve_pool`.
3709 pub async fn create(&self, mut instance: T) -> Result<T, crate::orm::write::WriteError>
3710 where
3711 T: serde::Serialize
3712 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3713 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
3714 + HydrateRelated,
3715 {
3716 use crate::orm::write::WriteError;
3717 let map = serialize_to_map(&instance)?;
3718
3719 // Same pre-DB validation pipeline the dynamic
3720 // `insert_json` path runs — choices + FK existence +
3721 // M2M shape. Empty-string + required-field checks are
3722 // intentionally relaxed on the typed path: a Rust
3723 // `pub title: String` field set to `""` is the caller's
3724 // deliberate choice, not a form-default leak, and
3725 // missing-required can't happen because the struct
3726 // forced the caller to supply every column at compile
3727 // time. We only validate the things the typed path
3728 // can't catch at compile time.
3729 let meta = crate::migrate::ModelMeta::for_::<T>();
3730 let validation_errors = crate::orm::validation::validate_on_typed_create(&meta, &map).await;
3731 if !validation_errors.is_empty() {
3732 return Err(WriteError::Multiple {
3733 errors: validation_errors,
3734 });
3735 }
3736
3737 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
3738 let backend = pool.backend_name();
3739 let stmt = build_insert_one_for::<T>(backend, &map)?;
3740 let atomic = self.should_atomic_wrap();
3741 // Post-execution SQL classification: turns the DB's
3742 // UNIQUE / FK / NOT NULL / CHECK violations into the
3743 // structured `WriteError` variants instead of a raw
3744 // `Sqlx(_)` 500. Symmetric with `DynQuerySet::insert_json`.
3745 match pool {
3746 DbPool::Sqlite(pool) => {
3747 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
3748 let row_result = if atomic {
3749 let mut tx = pool.begin().await.map_err(WriteError::Sqlx)?;
3750 let r = sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
3751 .fetch_one(&mut *tx)
3752 .await;
3753 match r {
3754 Ok(row) => {
3755 tx.commit().await.map_err(WriteError::Sqlx)?;
3756 Ok(row)
3757 }
3758 Err(e) => {
3759 let _ = tx.rollback().await;
3760 Err(e)
3761 }
3762 }
3763 } else {
3764 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
3765 .fetch_one(&pool)
3766 .await
3767 };
3768 let mut row = row_result.map_err(|e| {
3769 crate::orm::validation::classify_sql_error(&e, &map)
3770 .unwrap_or(WriteError::Sqlx(e))
3771 })?;
3772 // BUG-16 step 2: every materialised row, including the
3773 // post-INSERT readback, needs `parent_id` +
3774 // `junction_table` seeded on its M2M slots — otherwise
3775 // `row.tags.add(...)` is a silent no-op.
3776 row.set_m2m_parent_ids();
3777 // Carry form-staged M2M pending ids from the caller's
3778 // instance onto the readback row, then flush them to
3779 // junction rows now that parent_id + junction_table are
3780 // seeded on the readback row.
3781 instance.take_pending_m2m_into(&mut row);
3782 row.write_pending_m2m().await?;
3783 Ok(row)
3784 }
3785 DbPool::Postgres(pool) => {
3786 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
3787 let row_result = if atomic {
3788 let mut tx = pool.begin().await.map_err(WriteError::Sqlx)?;
3789 let r = sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
3790 .fetch_one(&mut *tx)
3791 .await;
3792 match r {
3793 Ok(row) => {
3794 tx.commit().await.map_err(WriteError::Sqlx)?;
3795 Ok(row)
3796 }
3797 Err(e) => {
3798 let _ = tx.rollback().await;
3799 Err(e)
3800 }
3801 }
3802 } else {
3803 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
3804 .fetch_one(&pool)
3805 .await
3806 };
3807 let mut row = row_result.map_err(|e| {
3808 crate::orm::validation::classify_sql_error(&e, &map)
3809 .unwrap_or(WriteError::Sqlx(e))
3810 })?;
3811 row.set_m2m_parent_ids();
3812 instance.take_pending_m2m_into(&mut row);
3813 row.write_pending_m2m().await?;
3814 Ok(row)
3815 }
3816 }
3817 }
3818
3819 /// INSERT many rows in a single statement. Returns the number of
3820 /// rows inserted. The full populated rows aren't materialised —
3821 /// use a follow-up `Model::objects().filter(...).fetch()` if you
3822 /// need them.
3823 ///
3824 /// Empty input is a no-op (returns Ok(0)) — the alternative
3825 /// (building an `INSERT INTO t () VALUES ()` and failing at the
3826 /// DB) doesn't help anyone.
3827 ///
3828 /// Fires `bulk_post_save:<table>` once with `{ ids, created: true,
3829 /// actor }` when at least one row was inserted. Per-row
3830 /// `pre_save` / `post_save` are NOT fired — use [`Self::save`]
3831 /// when per-row signal semantics are required.
3832 pub async fn bulk_create(&self, instances: Vec<T>) -> Result<u64, crate::orm::write::WriteError>
3833 where
3834 T: serde::Serialize,
3835 {
3836 use crate::orm::write::WriteError;
3837 if instances.is_empty() {
3838 return Ok(0);
3839 }
3840 let maps: Result<Vec<_>, _> = instances.iter().map(serialize_to_map).collect();
3841 let maps = maps?;
3842 // Validate every instance through the typed-create
3843 // pipeline. Collected into one `Multiple` so a caller
3844 // that submitted ten rows and got two bad ones can fix
3845 // both in one pass.
3846 let meta = crate::migrate::ModelMeta::for_::<T>();
3847 let mut all_errors: Vec<WriteError> = Vec::new();
3848 for map in &maps {
3849 let errs = crate::orm::validation::validate_on_typed_create(&meta, map).await;
3850 all_errors.extend(errs);
3851 }
3852 if !all_errors.is_empty() {
3853 return Err(WriteError::Multiple { errors: all_errors });
3854 }
3855 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
3856 let backend = pool.backend_name();
3857 let mut stmt = build_insert_many_for::<T>(backend, &maps)?;
3858 // First row's map is used to enrich UNIQUE / FK
3859 // messages with the offending value when the engine
3860 // doesn't name it. Imperfect for bulk (the failing row
3861 // could be later in the batch) but better than the raw
3862 // sqlx error.
3863 let first_map = maps.first().cloned().unwrap_or_default();
3864 // Add `RETURNING <pk>` so the bulk_post_save signal payload can
3865 // carry the inserted PKs. Both backends support this — SQLite
3866 // since 3.35, Postgres natively. Replaces the previous
3867 // `execute()` + rows_affected path; count comes from the
3868 // returned row vector instead.
3869 let pk = pk_field::<T>();
3870 if let Some(field) = pk {
3871 stmt.returning_col(Alias::new(field.name));
3872 }
3873 let atomic = self.should_atomic_wrap();
3874 let ids: Vec<JsonValue> = match pool {
3875 DbPool::Sqlite(pool) => {
3876 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
3877 let rows = if atomic {
3878 let mut tx = pool.begin().await.map_err(WriteError::Sqlx)?;
3879 let r = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3880 .fetch_all(&mut *tx)
3881 .await;
3882 match r {
3883 Ok(rows) => {
3884 tx.commit().await.map_err(WriteError::Sqlx)?;
3885 rows
3886 }
3887 Err(e) => {
3888 let _ = tx.rollback().await;
3889 return Err(crate::orm::validation::classify_sql_error(&e, &first_map)
3890 .unwrap_or(WriteError::Sqlx(e)));
3891 }
3892 }
3893 } else {
3894 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3895 .fetch_all(&pool)
3896 .await
3897 .map_err(|e| {
3898 crate::orm::validation::classify_sql_error(&e, &first_map)
3899 .unwrap_or(WriteError::Sqlx(e))
3900 })?
3901 };
3902 match pk {
3903 Some(field) => rows
3904 .iter()
3905 .map(|r| backend_sqlite::pk_to_json(r, field.name, field.ty))
3906 .collect::<Result<_, _>>()
3907 .map_err(WriteError::Sqlx)?,
3908 None => Vec::new(),
3909 }
3910 }
3911 DbPool::Postgres(pool) => {
3912 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
3913 let rows = if atomic {
3914 let mut tx = pool.begin().await.map_err(WriteError::Sqlx)?;
3915 let r = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3916 .fetch_all(&mut *tx)
3917 .await;
3918 match r {
3919 Ok(rows) => {
3920 tx.commit().await.map_err(WriteError::Sqlx)?;
3921 rows
3922 }
3923 Err(e) => {
3924 let _ = tx.rollback().await;
3925 return Err(crate::orm::validation::classify_sql_error(&e, &first_map)
3926 .unwrap_or(WriteError::Sqlx(e)));
3927 }
3928 }
3929 } else {
3930 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3931 .fetch_all(&pool)
3932 .await
3933 .map_err(|e| {
3934 crate::orm::validation::classify_sql_error(&e, &first_map)
3935 .unwrap_or(WriteError::Sqlx(e))
3936 })?
3937 };
3938 match pk {
3939 Some(field) => rows
3940 .iter()
3941 .map(|r| backend_pg::pk_to_json(r, field.name, field.ty))
3942 .collect::<Result<_, _>>()
3943 .map_err(WriteError::Sqlx)?,
3944 None => Vec::new(),
3945 }
3946 }
3947 };
3948 let count = ids.len() as u64;
3949 if !ids.is_empty() {
3950 crate::signals::emit_bulk_post_save::<T>(ids, true).await;
3951 }
3952 Ok(count)
3953 }
3954
3955 /// Django's `get_or_create`: fetch the first row matching `predicate`;
3956 /// if none exists, insert `defaults` and return it. Returns
3957 /// `(row, created)` so the caller can branch on whether the write
3958 /// happened. Two queries on the miss path (filter+first then create),
3959 /// one query on the hit path.
3960 ///
3961 /// ## Concurrency
3962 ///
3963 /// Convergent under concurrent callers: if two callers both miss the
3964 /// SELECT and race to INSERT, the one that loses gets a
3965 /// `UniqueViolation`; that error is caught here and the existing row
3966 /// is re-fetched, so both callers return the same row with
3967 /// `created = false` for the loser. A UNIQUE constraint on the
3968 /// predicate columns is required for true at-most-one semantics — the
3969 /// constraint is what makes the convergence deterministic.
3970 pub async fn get_or_create(
3971 &self,
3972 predicate: Predicate<T>,
3973 defaults: T,
3974 ) -> Result<(T, bool), crate::orm::write::WriteError>
3975 where
3976 T: serde::Serialize
3977 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3978 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
3979 + HydrateRelated,
3980 {
3981 use crate::orm::write::WriteError;
3982
3983 // Read-your-writes: probe for the existing row on the WRITE database,
3984 // not a (possibly lagging) read replica — otherwise a read/write-split
3985 // router could miss a just-written row and insert a duplicate. The
3986 // following `create()` already resolves the same write target.
3987 let write_pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
3988 if let Some(existing) = pin_to_pool(self.filter(predicate.clone()), &write_pool)
3989 .first()
3990 .await
3991 .map_err(WriteError::Sqlx)?
3992 {
3993 return Ok((existing, false));
3994 }
3995
3996 // Attempt the INSERT. On a UNIQUE violation (concurrent writer won the
3997 // race between our SELECT and this INSERT), catch the error and
3998 // re-SELECT to return the now-existing row with created=false.
3999 // This is the standard try-insert-then-fetch convergence pattern.
4000 //
4001 // Note on transaction semantics: a plain INSERT is atomic at the
4002 // statement level. Wrapping SELECT+INSERT in a serialisable transaction
4003 // would be stricter but requires SAVEPOINT support to recover from the
4004 // Postgres "aborted transaction" state after a constraint violation —
4005 // a per-operation SAVEPOINT would add two extra round-trips on every
4006 // write for marginal gain. The UNIQUE-constraint backstop plus this
4007 // re-fetch gives the same observable guarantee: callers always converge
4008 // on the same row and never see a spurious UniqueViolation.
4009 match self.create(defaults).await {
4010 Ok(created) => Ok((created, true)),
4011 Err(WriteError::UniqueViolation { .. }) => {
4012 // A concurrent writer inserted the row between our SELECT and
4013 // our INSERT. Re-fetch the now-existing row.
4014 let existing = pin_to_pool(self.filter(predicate), &write_pool)
4015 .first()
4016 .await
4017 .map_err(WriteError::Sqlx)?
4018 .ok_or_else(|| {
4019 WriteError::Sqlx(sqlx::Error::Protocol(
4020 "get_or_create: row vanished after UniqueViolation re-fetch"
4021 .to_string(),
4022 ))
4023 })?;
4024 Ok((existing, false))
4025 }
4026 Err(e) => Err(e),
4027 }
4028 }
4029
4030 /// Django's `update_or_create`: fetch the first row matching
4031 /// `predicate`; if found, update its non-PK columns with the
4032 /// `defaults` instance's values and return the fresh row;
4033 /// otherwise insert `defaults` and return it. Returns
4034 /// `(row, created)` so the caller can branch on the path taken.
4035 ///
4036 /// The defaults' PK is intentionally ignored on the update path —
4037 /// the matched row keeps its original PK. On the insert path the
4038 /// defaults' PK is honoured (autoincrement sentinel `0` → DB
4039 /// assigns; explicit value → DB uses it).
4040 ///
4041 /// ## Concurrency
4042 ///
4043 /// Convergent under concurrent callers: if two callers both miss the
4044 /// SELECT and race to INSERT, the loser gets a `UniqueViolation`; that
4045 /// error is caught here and the existing row is re-fetched, then the
4046 /// update is applied to it. Both callers converge on the same row, with
4047 /// `created = false` for the loser. A UNIQUE constraint on the predicate
4048 /// columns is required for deterministic convergence.
4049 ///
4050 /// Implementation: 2 queries on the hit path (`first` + UPDATE + re-fetch),
4051 /// 2 queries on the miss+create path (`first` + `create`), or
4052 /// 4 queries on the miss+race path (`first` + failed-INSERT + `first`
4053 /// + UPDATE + re-fetch).
4054 pub async fn update_or_create(
4055 &self,
4056 predicate: Predicate<T>,
4057 defaults: T,
4058 ) -> Result<(T, bool), crate::orm::write::WriteError>
4059 where
4060 T: serde::Serialize
4061 + Clone
4062 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4063 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
4064 + HydrateRelated,
4065 {
4066 use crate::orm::write::WriteError;
4067 let pk = pk_field::<T>().ok_or_else(|| {
4068 WriteError::Sqlx(sqlx::Error::Protocol(
4069 "update_or_create: model has no primary key".to_string(),
4070 ))
4071 })?;
4072 let pk_name = pk.name;
4073
4074 // Read-your-writes: the existence probe and the post-update re-fetch
4075 // run on the WRITE database, so a read/write-split router doesn't miss
4076 // a just-written row (duplicate insert) or read a stale row back. The
4077 // intervening UPDATE already routes to the same write target.
4078 let write_pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4079
4080 // Shared helper: given the existing row (already fetched) and the
4081 // defaults instance, apply the non-PK column update and return the
4082 // re-fetched row. Used by both the direct-hit path and the
4083 // UniqueViolation-convergence path so the update logic is in one place.
4084 macro_rules! do_update {
4085 ($existing:expr, $defaults:expr) => {{
4086 let existing: T = $existing;
4087 let defaults: T = $defaults;
4088
4089 // Serialize defaults, drop the PK so the matched row's PK is
4090 // preserved, then UPDATE WHERE <pk_col> = <existing_pk>.
4091 let mut update_map = serialize_to_map(&defaults)?;
4092 update_map.remove(pk_name);
4093
4094 // Build a PK predicate from the existing row's serialized PK
4095 // value. Goes through serde_json so any PK type (i64, String,
4096 // Uuid) round-trips correctly through sea-query.
4097 let existing_json =
4098 serde_json::to_value(&existing).map_err(WriteError::SerializeFailed)?;
4099 let pk_value_json = existing_json
4100 .get(pk_name)
4101 .cloned()
4102 .unwrap_or(serde_json::Value::Null);
4103 let pk_sea = crate::orm::write::json_to_sea_value(
4104 pk.ty,
4105 &pk_value_json,
4106 false,
4107 pk_name,
4108 None,
4109 )?;
4110 let pk_pred: Predicate<T> = Predicate::new(
4111 sea_query::Expr::col(sea_query::Alias::new(pk_name)).eq(pk_sea),
4112 );
4113
4114 // Run the UPDATE.
4115 self.filter(pk_pred).update_values(update_map).await?;
4116
4117 // Re-fetch to return the populated row.
4118 let pk_sea2 = crate::orm::write::json_to_sea_value(
4119 pk.ty,
4120 &pk_value_json,
4121 false,
4122 pk_name,
4123 None,
4124 )?;
4125 let refetch_pred: Predicate<T> = Predicate::new(
4126 sea_query::Expr::col(sea_query::Alias::new(pk_name)).eq(pk_sea2),
4127 );
4128 pin_to_pool(self.filter(refetch_pred), &write_pool)
4129 .first()
4130 .await
4131 .map_err(WriteError::Sqlx)?
4132 .ok_or_else(|| {
4133 WriteError::Sqlx(sqlx::Error::Protocol(
4134 "update_or_create: row vanished between UPDATE and re-fetch"
4135 .to_string(),
4136 ))
4137 })?
4138 }};
4139 }
4140
4141 if let Some(existing) = pin_to_pool(self.filter(predicate.clone()), &write_pool)
4142 .first()
4143 .await
4144 .map_err(WriteError::Sqlx)?
4145 {
4146 let updated = do_update!(existing, defaults);
4147 return Ok((updated, false));
4148 }
4149
4150 // Attempt the INSERT. On a UNIQUE violation (concurrent writer won the
4151 // race between our SELECT and this INSERT), catch the error, re-fetch
4152 // the now-existing row, and apply the update to it — same convergence
4153 // as get_or_create but with an extra UPDATE step.
4154 match self.create(defaults.clone()).await {
4155 Ok(created) => Ok((created, true)),
4156 Err(WriteError::UniqueViolation { .. }) => {
4157 // A concurrent writer inserted the row between our SELECT and
4158 // our INSERT. Re-fetch then update, same as the direct-hit path.
4159 let existing = pin_to_pool(self.filter(predicate), &write_pool)
4160 .first()
4161 .await
4162 .map_err(WriteError::Sqlx)?
4163 .ok_or_else(|| {
4164 WriteError::Sqlx(sqlx::Error::Protocol(
4165 "update_or_create: row vanished after UniqueViolation re-fetch"
4166 .to_string(),
4167 ))
4168 })?;
4169 let updated = do_update!(existing, defaults);
4170 Ok((updated, false))
4171 }
4172 Err(e) => Err(e),
4173 }
4174 }
4175
4176 /// INSERT-or-UPDATE keyed on the primary key. The row's PK column
4177 /// is the conflict target; on a hit, every non-PK column is
4178 /// overwritten with the supplied value. Returns the row as the DB
4179 /// stored it (post-upsert).
4180 ///
4181 /// Both backends use `INSERT ... ON CONFLICT(<pk>) DO UPDATE SET
4182 /// col = excluded.col, ...`. The SQLite and Postgres syntax happens
4183 /// to match exactly here so a single sea-query `OnConflict` builder
4184 /// covers both.
4185 pub async fn upsert(&self, instance: T) -> Result<T, crate::orm::write::WriteError>
4186 where
4187 T: serde::Serialize
4188 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4189 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4190 {
4191 let map = serialize_to_map(&instance)?;
4192 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4193 let backend = pool.backend_name();
4194 let mut stmt = build_insert_one_for::<T>(backend, &map)?;
4195
4196 // Conflict target = PK column. update_columns = every non-PK
4197 // column the body included. sea-query renders `DO UPDATE SET
4198 // col = excluded.col` (SQLite) / `DO UPDATE SET col =
4199 // EXCLUDED.col` (PG) — both forms work cross-dialect.
4200 let pk_name = T::FIELDS
4201 .iter()
4202 .find(|f| f.primary_key)
4203 .map(|f| f.name)
4204 .ok_or_else(|| {
4205 crate::orm::write::WriteError::Sqlx(sqlx::Error::Protocol(
4206 "upsert: model has no primary key — use get_or_create or create instead"
4207 .to_string(),
4208 ))
4209 })?;
4210 let update_cols: Vec<Alias> = T::FIELDS
4211 .iter()
4212 .filter(|f| !f.primary_key && map.contains_key(f.name))
4213 .map(|f| Alias::new(f.name))
4214 .collect();
4215 let mut on_conflict = sea_query::OnConflict::column(Alias::new(pk_name));
4216 if !update_cols.is_empty() {
4217 on_conflict.update_columns(update_cols);
4218 } else {
4219 // No non-PK columns to overwrite — this is a "INSERT OR
4220 // IGNORE" shape. sea-query encodes that as `DO NOTHING`.
4221 on_conflict.do_nothing();
4222 }
4223 stmt.on_conflict(on_conflict);
4224
4225 match pool {
4226 DbPool::Sqlite(pool) => {
4227 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4228 let row = sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
4229 .fetch_one(&pool)
4230 .await?;
4231 Ok(row)
4232 }
4233 DbPool::Postgres(pool) => {
4234 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4235 let row = sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
4236 .fetch_one(&pool)
4237 .await?;
4238 Ok(row)
4239 }
4240 }
4241 }
4242
4243 /// `create` against an explicit Postgres pool. The Postgres
4244 /// counterpart of [`Self::create`] for models with Postgres-only
4245 /// field types (Array, Inet, MacAddr, FullText), whose `FromRow`
4246 /// impl exists only for `PgRow`.
4247 pub async fn create_pg(
4248 &self,
4249 instance: T,
4250 pool: &sqlx::PgPool,
4251 ) -> Result<T, crate::orm::write::WriteError>
4252 where
4253 T: serde::Serialize + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4254 {
4255 let map = serialize_to_map(&instance)?;
4256 let stmt = build_insert_one_for::<T>("postgres", &map)?;
4257 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4258 let row = sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
4259 .fetch_one(pool)
4260 .await?;
4261 Ok(row)
4262 }
4263
4264 /// Apply per-row differing values to a list of instances in one
4265 /// statement. Each instance carries its own PK and its own
4266 /// column values; the generated SQL uses one `CASE pk WHEN ...
4267 /// THEN ... END` per non-PK column, plus a `WHERE pk IN (...)`
4268 /// to scope the update.
4269 ///
4270 /// Mirrors Django's `QuerySet.bulk_update(objs, fields)` but
4271 /// updates every non-PK column rather than asking the caller to
4272 /// list them. Returns the number of rows affected.
4273 ///
4274 /// Empty input is a no-op (returns 0). The pattern works on both
4275 /// SQLite and Postgres — the `CASE` expression is SQL-standard.
4276 ///
4277 /// Limitations:
4278 /// - All instances must have a non-default PK (the caller has
4279 /// already loaded the rows). Default-PK instances are skipped.
4280 /// - Bulk-write signals are NOT fired by this path — it's the
4281 /// `Manager::bulk_create` analogue for UPDATE, deliberately
4282 /// silent for speed.
4283 pub async fn bulk_update(&self, instances: Vec<T>) -> Result<u64, crate::orm::write::WriteError>
4284 where
4285 T: serde::Serialize,
4286 {
4287 use crate::orm::write::{WriteError, is_default_pk, json_to_sea_value};
4288 if instances.is_empty() {
4289 return Ok(0);
4290 }
4291 let pk = pk_field::<T>().ok_or_else(|| {
4292 WriteError::Sqlx(sqlx::Error::Protocol(
4293 "bulk_update: model has no primary key".to_string(),
4294 ))
4295 })?;
4296 let pk_name = pk.name;
4297 let pk_ty = pk.ty;
4298
4299 // Serialize every instance, collecting (pk_value, full_map)
4300 // for the CASE branches and the IN clause. Skip rows whose
4301 // PK is still the default sentinel — they were never
4302 // persisted and a bulk UPDATE on them is a no-op anyway.
4303 let mut serialized: Vec<(
4304 serde_json::Value,
4305 serde_json::Map<String, serde_json::Value>,
4306 )> = Vec::with_capacity(instances.len());
4307 for instance in &instances {
4308 let map = serialize_to_map(instance)?;
4309 let pk_val = map.get(pk_name).cloned().unwrap_or(serde_json::Value::Null);
4310 if is_default_pk(pk_ty, &pk_val) {
4311 continue;
4312 }
4313 serialized.push((pk_val, map));
4314 }
4315 if serialized.is_empty() {
4316 return Ok(0);
4317 }
4318
4319 // Collect the list of non-PK columns to update from the
4320 // first row (every row contributes the same column set
4321 // because they're typed instances of T).
4322 let update_cols: Vec<&crate::orm::FieldSpec> =
4323 T::FIELDS.iter().filter(|f| !f.primary_key).collect();
4324
4325 // Build the UPDATE: one CASE per column, IN clause for the
4326 // WHERE. Goes through sea-query's update statement for
4327 // backend portability.
4328 let mut stmt = sea_query::Query::update();
4329 stmt.table(crate::db::router::schema_qualified_table(T::TABLE));
4330
4331 for field in &update_cols {
4332 // CASE pk_col
4333 // WHEN <pk1> THEN <val1>
4334 // WHEN <pk2> THEN <val2>
4335 // ...
4336 // END
4337 let mut case = sea_query::CaseStatement::new();
4338 for (pk_val, map) in &serialized {
4339 let val = map
4340 .get(field.name)
4341 .cloned()
4342 .unwrap_or(serde_json::Value::Null);
4343 let cell = json_to_sea_value(
4344 field.ty,
4345 &val,
4346 field.nullable,
4347 field.name,
4348 fk_pk_hint(field),
4349 )?;
4350 let pk_sea = json_to_sea_value(pk_ty, pk_val, false, pk_name, None)?;
4351 case = case.case(sea_query::Expr::col(Alias::new(pk_name)).eq(pk_sea), cell);
4352 }
4353 stmt.value(Alias::new(field.name), case);
4354 }
4355
4356 // WHERE pk IN (<pk1>, <pk2>, ...)
4357 let pk_seas: Vec<sea_query::Value> = serialized
4358 .iter()
4359 .map(|(pk_val, _)| json_to_sea_value(pk_ty, pk_val, false, pk_name, None))
4360 .collect::<Result<_, _>>()?;
4361 stmt.and_where(sea_query::Expr::col(Alias::new(pk_name)).is_in(pk_seas));
4362
4363 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4364 let affected = match pool {
4365 DbPool::Sqlite(pool) => {
4366 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4367 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
4368 .execute(&pool)
4369 .await
4370 .map_err(WriteError::Sqlx)?
4371 .rows_affected()
4372 }
4373 DbPool::Postgres(pool) => {
4374 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4375 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
4376 .execute(&pool)
4377 .await
4378 .map_err(WriteError::Sqlx)?
4379 .rows_affected()
4380 }
4381 };
4382 Ok(affected)
4383 }
4384
4385 /// Run a hand-written SQL query and return typed `Vec<T>` rows.
4386 ///
4387 /// The escape hatch for queries the QuerySet builder can't (or
4388 /// shouldn't) model — CTEs, vendor-specific functions, ad-hoc
4389 /// reporting. Delegates to `sqlx::query_as` against the ambient
4390 /// pool and dispatches on backend, so user code stays portable.
4391 /// The string is sent verbatim; no parameter binding (use
4392 /// `Predicate` / the typed query path for parameterised
4393 /// queries). Inject input only after manual sanitisation.
4394 ///
4395 /// Skips the `select_related` / `prefetch_related` chain — those
4396 /// only apply to the typed QuerySet build path.
4397 pub async fn raw(&self, sql: &str) -> Result<Vec<T>, sqlx::Error>
4398 where
4399 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4400 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4401 {
4402 // Routing: `raw` resolves to the READ database (most raw statements
4403 // are SELECTs, which this returns as `Vec<T>`). Under a read/write-
4404 // split router, a raw statement that WRITES must pin the write pool
4405 // explicitly via `.on(&pool)` / `.on_pg(&pool)`, since the router
4406 // cannot inspect arbitrary SQL to know it mutates.
4407 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Read);
4408 match pool {
4409 DbPool::Sqlite(pool) => {
4410 sqlx::query_as::<sqlx::Sqlite, T>(sql)
4411 .fetch_all(&pool)
4412 .await
4413 }
4414 DbPool::Postgres(pool) => {
4415 sqlx::query_as::<sqlx::Postgres, T>(sql)
4416 .fetch_all(&pool)
4417 .await
4418 }
4419 }
4420 }
4421
4422 /// `bulk_create` against an explicit Postgres pool.
4423 pub async fn bulk_create_pg(
4424 &self,
4425 instances: Vec<T>,
4426 pool: &sqlx::PgPool,
4427 ) -> Result<u64, crate::orm::write::WriteError>
4428 where
4429 T: serde::Serialize,
4430 {
4431 if instances.is_empty() {
4432 return Ok(0);
4433 }
4434 let maps: Result<Vec<_>, _> = instances.iter().map(serialize_to_map).collect();
4435 let maps = maps?;
4436 let stmt = build_insert_many_for::<T>("postgres", &maps)?;
4437 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4438 let result = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
4439 .execute(pool)
4440 .await?;
4441 Ok(result.rows_affected())
4442 }
4443}
4444
4445// QuerySetTx (struct + impl) moved to `super::tx`; re-exported above.
4446
4447impl<T: Model> Manager<T> {
4448 /// Begin a new query on this manager attached to the given open transaction.
4449 ///
4450 /// Sugar for `T::objects().on_tx(tx)` — lets callers skip the intermediate
4451 /// `QuerySet` construction when they want to go straight to a terminal:
4452 ///
4453 /// ```rust,ignore
4454 /// umbral::db::transaction(|tx| async move {
4455 /// let post = Post::objects().on_tx(tx).create(new_post).await?;
4456 /// Ok::<_, MyError>(post)
4457 /// }).await?;
4458 /// ```
4459 pub fn on_tx<'a>(&self, tx: &'a mut crate::db::Transaction) -> QuerySetTx<'a, T> {
4460 self.queryset().on_tx(tx)
4461 }
4462
4463 /// INSERT one row inside `tx` and return the populated row.
4464 ///
4465 /// This is the primary Manager-level entry point for transactional writes.
4466 /// Equivalent to `Post::objects().on_tx(tx).create(instance)` but more
4467 /// ergonomic when you only need the one INSERT (no filter chain needed).
4468 ///
4469 /// ```rust,ignore
4470 /// umbral::db::transaction(|tx| async move {
4471 /// let post = Post::objects().create_in_tx(new_post, tx).await?;
4472 /// Ok::<_, MyError>(post)
4473 /// }).await?;
4474 /// ```
4475 pub async fn create_in_tx(
4476 &self,
4477 instance: T,
4478 tx: &mut crate::db::Transaction,
4479 ) -> Result<T, crate::orm::write::WriteError>
4480 where
4481 T: serde::Serialize
4482 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4483 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
4484 + HydrateRelated,
4485 {
4486 let map = serialize_to_map(&instance)?;
4487 let stmt = build_insert_one_for::<T>(tx.backend_name(), &map)?;
4488 match tx.backend_name() {
4489 "sqlite" => {
4490 let inner = tx.as_sqlite_mut().unwrap();
4491 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4492 let mut row = sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
4493 .fetch_one(&mut **inner)
4494 .await?;
4495 row.set_m2m_parent_ids();
4496 Ok(row)
4497 }
4498 _ => {
4499 let inner = tx.as_pg_mut().unwrap();
4500 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4501 let mut row = sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
4502 .fetch_one(&mut **inner)
4503 .await?;
4504 row.set_m2m_parent_ids();
4505 Ok(row)
4506 }
4507 }
4508 }
4509
4510 /// INSERT many rows inside `tx`.
4511 ///
4512 /// Returns the number of rows inserted. Empty input is a no-op.
4513 pub async fn bulk_create_in_tx(
4514 &self,
4515 instances: Vec<T>,
4516 tx: &mut crate::db::Transaction,
4517 ) -> Result<u64, crate::orm::write::WriteError>
4518 where
4519 T: serde::Serialize,
4520 {
4521 if instances.is_empty() {
4522 return Ok(0);
4523 }
4524 let maps: Result<Vec<_>, _> = instances.iter().map(serialize_to_map).collect();
4525 let maps = maps?;
4526 let stmt = build_insert_many_for::<T>(tx.backend_name(), &maps)?;
4527 match tx.backend_name() {
4528 "sqlite" => {
4529 let inner = tx.as_sqlite_mut().unwrap();
4530 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4531 let result = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
4532 .execute(&mut **inner)
4533 .await?;
4534 Ok(result.rows_affected())
4535 }
4536 _ => {
4537 let inner = tx.as_pg_mut().unwrap();
4538 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4539 let result = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
4540 .execute(&mut **inner)
4541 .await?;
4542 Ok(result.rows_affected())
4543 }
4544 }
4545 }
4546
4547 // =====================================================================
4548 // Per-instance signal-firing write methods.
4549 //
4550 // `save(instance)` and `delete_instance(instance)` are the methods that
4551 // fire the ORM lifecycle signals (`pre_save` / `post_save` /
4552 // `pre_delete` / `post_delete`). The existing bulk methods
4553 // (`create`, `bulk_create`, `QuerySet::update_values`,
4554 // `QuerySet::delete`) remain signal-free, matching Django's own
4555 // behaviour: bulk operations bypass signals for performance.
4556 //
4557 // Signal name format: `<event>:<table>` — e.g. `post_save:post`.
4558 // Payload shapes:
4559 // save: `{ "instance": <M as JSON>, "created": bool }`
4560 // delete: `{ "instance": <M as JSON> }`
4561 //
4562 // The `created` flag on save follows Django's convention:
4563 // `true` when the PK is the default sentinel → INSERT path.
4564 // `false` when the PK is non-default → UPDATE path.
4565 // =====================================================================
4566
4567 /// Save one instance, firing `pre_save` + `post_save` signals.
4568 ///
4569 /// Determines INSERT vs UPDATE by checking whether the primary key
4570 /// is the autoincrement sentinel (`0` for integers, nil UUID, empty
4571 /// string). If it is, an INSERT is performed (`created = true`);
4572 /// otherwise an `UPDATE ... WHERE pk = <value>` is run (`created = false`).
4573 ///
4574 /// Returns the row as it exists in the database after the write
4575 /// (populated PK for inserts, same row for updates).
4576 ///
4577 /// ## Signal contract
4578 ///
4579 /// - `pre_save:<table>` fires before the database write with
4580 /// `{ "instance": ..., "created": bool, "actor": ... }`.
4581 /// - `post_save:<table>` fires after the database write with the
4582 /// DB-read-back row and the same envelope keys.
4583 ///
4584 /// The `"actor"` value is set by the nearest enclosing
4585 /// [`crate::signals::with_actor`] scope; `Value::Null` when no
4586 /// scope is active.
4587 ///
4588 /// ## Bulk paths fire bulk signals, not per-row signals
4589 ///
4590 /// `Manager::create`, `Manager::bulk_create`, and
4591 /// `QuerySet::update_values` / `QuerySet::delete` do NOT fire
4592 /// per-row `post_save` / `post_delete`. They fire
4593 /// `bulk_post_save:<table>` / `bulk_post_delete:<table>` once per
4594 /// statement with the affected PKs in the payload. Use `save` /
4595 /// `delete_instance` when per-row signal semantics are needed.
4596 pub async fn save(&self, instance: T) -> Result<T, crate::orm::write::SaveError>
4597 where
4598 T: serde::Serialize
4599 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4600 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4601 {
4602 use crate::orm::write::{SaveError, is_default_pk};
4603 // Determine INSERT vs UPDATE by inspecting the PK field.
4604 let pk_field = T::FIELDS
4605 .iter()
4606 .find(|f| f.primary_key)
4607 .ok_or(SaveError::NoPrimaryKey)?;
4608 let map = serialize_to_map(&instance).map_err(SaveError::Write)?;
4609 let pk_val = map
4610 .get(pk_field.name)
4611 .cloned()
4612 .unwrap_or(serde_json::Value::Null);
4613 let created = is_default_pk(pk_field.ty, &pk_val);
4614
4615 // Fire pre_save before the write.
4616 crate::signals::emit_pre_save::<T>(&instance, created).await;
4617
4618 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4619 let backend = pool.backend_name();
4620
4621 if created {
4622 // INSERT path.
4623 let stmt = build_insert_one_for::<T>(backend, &map).map_err(SaveError::Write)?;
4624 let row = match pool {
4625 DbPool::Sqlite(pool) => {
4626 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4627 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
4628 .fetch_one(&pool)
4629 .await
4630 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
4631 }
4632 DbPool::Postgres(pool) => {
4633 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4634 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
4635 .fetch_one(&pool)
4636 .await
4637 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
4638 }
4639 };
4640 // Fire post_save with the DB-populated row.
4641 crate::signals::emit_post_save::<T>(&row, true).await;
4642 Ok(row)
4643 } else {
4644 // UPDATE path: UPDATE ... WHERE <pk> = <value> RETURNING *.
4645 use sea_query::{Alias, Expr, Query};
4646
4647 // gaps2 #92 — snapshot the pre-UPDATE row for `pre_update` /
4648 // `post_update` subscribers, but ONLY when one exists. The
4649 // extra SELECT-by-PK is gated on `has_subscribers` so the
4650 // common UPDATE path (no `*_update` listener) pays nothing.
4651 // Best-effort TOCTOU: the snapshot reads the row before the
4652 // UPDATE; a concurrent writer between the two is accepted.
4653 let pre_table = T::TABLE;
4654 let want_pre = crate::signals::has_subscribers(&format!("pre_update:{pre_table}"));
4655 let want_post = crate::signals::has_subscribers(&format!("post_update:{pre_table}"));
4656 let previous: Option<T> = if want_pre || want_post {
4657 let mut sel = Query::select();
4658 sel.from(crate::db::router::schema_qualified_table(T::TABLE));
4659 for field in T::FIELDS {
4660 sel.column(Alias::new(field.name));
4661 }
4662 let pk_sea_sel = crate::orm::write::json_to_sea_value(
4663 pk_field.ty,
4664 &pk_val,
4665 false,
4666 pk_field.name,
4667 None,
4668 )
4669 .map_err(SaveError::Write)?;
4670 sel.and_where(Expr::col(Alias::new(pk_field.name)).eq(pk_sea_sel));
4671 sel.limit(1);
4672 match pool {
4673 DbPool::Sqlite(ref pool) => {
4674 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
4675 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
4676 .fetch_optional(pool)
4677 .await
4678 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
4679 }
4680 DbPool::Postgres(ref pool) => {
4681 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
4682 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
4683 .fetch_optional(pool)
4684 .await
4685 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
4686 }
4687 }
4688 } else {
4689 None
4690 };
4691 // Fire pre_update before the UPDATE when both a snapshot exists
4692 // and a subscriber wants it.
4693 if want_pre {
4694 if let Some(prev) = &previous {
4695 crate::signals::emit_pre_update::<T>(prev, &instance).await;
4696 }
4697 }
4698
4699 let mut stmt = Query::update();
4700 stmt.table(crate::db::router::schema_qualified_table(T::TABLE));
4701 for field in T::FIELDS {
4702 if field.primary_key {
4703 continue;
4704 }
4705 let val = map
4706 .get(field.name)
4707 .cloned()
4708 .unwrap_or(serde_json::Value::Null);
4709 let sea_val = crate::orm::write::json_to_sea_value(
4710 field.ty,
4711 &val,
4712 field.nullable,
4713 field.name,
4714 fk_pk_hint(field),
4715 )
4716 .map_err(SaveError::Write)?;
4717 stmt.value(Alias::new(field.name), sea_val);
4718 }
4719 // WHERE pk = <value>
4720 let pk_sea = crate::orm::write::json_to_sea_value(
4721 pk_field.ty,
4722 &pk_val,
4723 false,
4724 pk_field.name,
4725 None,
4726 )
4727 .map_err(SaveError::Write)?;
4728 stmt.and_where(Expr::col(Alias::new(pk_field.name)).eq(pk_sea));
4729 // RETURNING * so we can return the updated row.
4730 stmt.returning_all();
4731
4732 let row = match pool {
4733 DbPool::Sqlite(pool) => {
4734 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4735 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
4736 .fetch_one(&pool)
4737 .await
4738 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
4739 }
4740 DbPool::Postgres(pool) => {
4741 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4742 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
4743 .fetch_one(&pool)
4744 .await
4745 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
4746 }
4747 };
4748 // Fire post_save with created=false.
4749 crate::signals::emit_post_save::<T>(&row, false).await;
4750 // gaps2 #92 — post_update carries the pre-UPDATE snapshot (old)
4751 // and the freshly-written row (new). Only when a subscriber
4752 // exists and the snapshot was captured.
4753 if want_post {
4754 if let Some(prev) = &previous {
4755 crate::signals::emit_post_update::<T>(prev, &row).await;
4756 }
4757 }
4758 Ok(row)
4759 }
4760 }
4761
4762 /// Delete one instance by primary key, firing `pre_delete` +
4763 /// `post_delete` signals.
4764 ///
4765 /// Issues `DELETE FROM <table> WHERE <pk> = <value>`. Returns the
4766 /// number of rows affected (0 if the row was already gone, 1 otherwise).
4767 ///
4768 /// ## Signal contract
4769 ///
4770 /// - `pre_delete:<table>` fires before the DELETE with
4771 /// `{ "instance": ..., "actor": ... }`.
4772 /// - `post_delete:<table>` fires after the DELETE with the same
4773 /// payload shape.
4774 ///
4775 /// The `"actor"` value is set by the nearest enclosing
4776 /// [`crate::signals::with_actor`] scope; `Value::Null` when no
4777 /// scope is active. The instance value passed to both signals is
4778 /// the value supplied by the caller — not a DB read-back. If you
4779 /// need the freshest DB state before deletion, fetch it first with
4780 /// `.get(...)` then pass to this method.
4781 ///
4782 /// ## Bulk paths fire bulk signals
4783 ///
4784 /// `QuerySet::delete()` (the filter-chain DELETE) fires
4785 /// `bulk_post_delete:<table>` with the list of affected PKs, not
4786 /// per-row `pre_delete` / `post_delete`. Use `delete_instance` for
4787 /// per-row signal semantics.
4788 pub async fn delete_instance(&self, instance: &T) -> Result<u64, crate::orm::write::SaveError>
4789 where
4790 T: serde::Serialize,
4791 {
4792 use crate::orm::write::SaveError;
4793 let pk_field = T::FIELDS
4794 .iter()
4795 .find(|f| f.primary_key)
4796 .ok_or(SaveError::NoPrimaryKey)?;
4797 let map = serialize_to_map(instance).map_err(SaveError::Write)?;
4798 let pk_val = map
4799 .get(pk_field.name)
4800 .cloned()
4801 .unwrap_or(serde_json::Value::Null);
4802
4803 // Fire pre_delete before the write.
4804 crate::signals::emit_pre_delete::<T>(instance).await;
4805
4806 let pk_sea =
4807 crate::orm::write::json_to_sea_value(pk_field.ty, &pk_val, false, pk_field.name, None)
4808 .map_err(SaveError::Write)?;
4809
4810 use sea_query::{Alias, Expr, Query};
4811 // Feature #72 — soft-delete redirect. For models tagged
4812 // `#[umbral(soft_delete)]`, set deleted_at instead of issuing
4813 // DELETE. Pre/post_delete signals still fire because the
4814 // logical contract ("this row is gone from the visible
4815 // table") is preserved — only the physical SQL changed.
4816 // Hard-delete is not exposed through delete_instance (it's
4817 // a typed per-row helper); call `QuerySet::filter(pk =
4818 // instance.id).hard_delete().delete()` when you need it.
4819 let stmt_sql = if T::SOFT_DELETE {
4820 let now = chrono::Utc::now();
4821 let mut up = Query::update();
4822 up.table(crate::db::router::schema_qualified_table(T::TABLE));
4823 up.value(
4824 Alias::new("deleted_at"),
4825 sea_query::Value::ChronoDateTimeUtc(Some(Box::new(now))),
4826 );
4827 up.and_where(Expr::col(Alias::new(pk_field.name)).eq(pk_sea));
4828 // Idempotency guard — don't bump an already-set timestamp.
4829 up.and_where(Expr::col(Alias::new("deleted_at")).is_null());
4830 SoftOrHardStatement::Update(up)
4831 } else {
4832 let mut stmt = Query::delete();
4833 stmt.from_table(crate::db::router::schema_qualified_table(T::TABLE));
4834 stmt.and_where(Expr::col(Alias::new(pk_field.name)).eq(pk_sea));
4835 SoftOrHardStatement::Delete(stmt)
4836 };
4837
4838 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4839 let affected = match (&pool, stmt_sql) {
4840 (DbPool::Sqlite(pool), SoftOrHardStatement::Delete(stmt)) => {
4841 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4842 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
4843 .execute(pool)
4844 .await
4845 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
4846 .rows_affected()
4847 }
4848 (DbPool::Postgres(pool), SoftOrHardStatement::Delete(stmt)) => {
4849 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4850 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
4851 .execute(pool)
4852 .await
4853 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
4854 .rows_affected()
4855 }
4856 (DbPool::Sqlite(pool), SoftOrHardStatement::Update(stmt)) => {
4857 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4858 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
4859 .execute(pool)
4860 .await
4861 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
4862 .rows_affected()
4863 }
4864 (DbPool::Postgres(pool), SoftOrHardStatement::Update(stmt)) => {
4865 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4866 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
4867 .execute(pool)
4868 .await
4869 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
4870 .rows_affected()
4871 }
4872 };
4873
4874 // Fire post_delete after the write.
4875 crate::signals::emit_post_delete::<T>(instance).await;
4876
4877 Ok(affected)
4878 }
4879}
4880
4881/// Internal enum used by `Manager::delete_instance` to dispatch on
4882/// soft-delete vs hard-delete without duplicating the four backend ×
4883/// statement match arms inline. One variant per SQL shape.
4884enum SoftOrHardStatement {
4885 Delete(sea_query::DeleteStatement),
4886 Update(sea_query::UpdateStatement),
4887}
4888
4889// Hydration helpers (hydrate_select_related, hydrate_select_related_nested,
4890// hydrate_prefetch_related, hydrate_reverse_fk_for_field, fetch_related_as_json,
4891// fetch_reverse_fk_children) moved to `super::hydration`.
4892
4893// Insert builders (serialize_to_map, build_insert_one_for, build_insert_many_for)
4894// and pk_field moved to `super::write_helpers`.
4895
4896// `decode_agg_sqlite` / `decode_agg_pg` moved to
4897// `backend_sqlite::decode_agg` / `backend_pg::decode_agg`.
4898
4899// =========================================================================
4900// #113: M2M-via-JOIN dedup decoders
4901//
4902// When .join_related() includes one or more M2M fields, the result
4903// set has one row per (parent, child) combo, so a parent with N
4904// matching children appears N times. The caller wants ONE T per
4905// parent with the M2M slot populated.
4906//
4907// The algorithm:
4908// - First time we see a parent PK: decode T via FromRow, hydrate
4909// any FK joins from this row.
4910// - Every subsequent row for the same parent: extract the M2M
4911// child JsonValue (or skip on LEFT JOIN miss).
4912// - Dedup children by (parent_pk, field, child_pk) so that
4913// joining TWO M2Ms doesn't multiply the child sets (the JOIN
4914// produces parent × m2m1 × m2m2 rows).
4915// - Hand each parent its M2M buckets via set_m2m_resolved_json.
4916// =========================================================================
4917
4918fn dedup_decode_sqlite<T: Model + HydrateRelated>(
4919 raw_rows: &[sqlx::sqlite::SqliteRow],
4920 fk_join_fields: &[String],
4921 m2m_join_fields: &[String],
4922) -> Result<Vec<T>, sqlx::Error>
4923where
4924 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>,
4925{
4926 // PK-agnostic dedup: read the parent PK back through the shape-aware
4927 // decoder (using the parent model's PK SqlType) and key by `pk_key`,
4928 // so i64 / String / Uuid parents all dedup correctly.
4929 let parent_pk_col = crate::migrate::ModelMeta::for_::<T>()
4930 .fields
4931 .into_iter()
4932 .find(|c| c.primary_key)
4933 .ok_or_else(|| {
4934 sqlx::Error::Protocol(format!(
4935 "umbral::orm::join_related: model `{}` has no primary key, M2M JOIN \
4936 dedup requires one",
4937 T::NAME
4938 ))
4939 })?;
4940 let registered = crate::migrate::registered_models();
4941 let mut typed: Vec<T> = Vec::new();
4942 let mut idx_by_pk: HashMap<String, usize> = HashMap::new();
4943 // (parent_pk_key, field) → Vec<JsonValue> + a Set of seen child PK keys.
4944 let mut buckets: HashMap<(String, String), Vec<JsonValue>> = HashMap::new();
4945 let mut seen_children: HashMap<(String, String), std::collections::HashSet<String>> =
4946 HashMap::new();
4947 for row in raw_rows {
4948 let Ok(parent_json) = crate::orm::dynamic::decode_to_json(row, &parent_pk_col) else {
4949 continue;
4950 };
4951 let parent_key = crate::orm::pk_key(&parent_json);
4952 if let std::collections::hash_map::Entry::Vacant(e) = idx_by_pk.entry(parent_key.clone()) {
4953 let mut t = <T as sqlx::FromRow<_>>::from_row(row)?;
4954 backend_sqlite::hydrate_joined_rels::<T>(&mut t, row, fk_join_fields)?;
4955 e.insert(typed.len());
4956 typed.push(t);
4957 }
4958 for m2m_field in m2m_join_fields {
4959 // The M2M slot is keyed by the FIRST segment (the M2M field
4960 // name); the full `m2m_field` path may carry an onward FK
4961 // chain (`"tags__category"`) the child decoder nests.
4962 let m2m_seg = m2m_field.split("__").next().unwrap_or(m2m_field.as_str());
4963 let Some(rel) = T::M2M_RELATIONS.iter().find(|r| r.field_name == m2m_seg) else {
4964 continue;
4965 };
4966 let Some(child_meta) = registered.iter().find(|m| m.table == rel.target_table) else {
4967 continue;
4968 };
4969 let Some(child_json) =
4970 backend_sqlite::extract_m2m_child_json::<T>(row, m2m_field, child_meta)?
4971 else {
4972 continue;
4973 };
4974 // Dedup by child PK (PK-agnostic) so multi-M2M cartesian
4975 // doesn't duplicate this field's children.
4976 let child_key = child_json
4977 .as_object()
4978 .and_then(|m| {
4979 let pk_col = child_meta.fields.iter().find(|c| c.primary_key)?;
4980 m.get(&pk_col.name).map(crate::orm::pk_key)
4981 })
4982 .unwrap_or_default();
4983 let key = (parent_key.clone(), m2m_seg.to_string());
4984 let seen = seen_children.entry(key.clone()).or_default();
4985 if seen.insert(child_key) {
4986 buckets.entry(key).or_default().push(child_json);
4987 }
4988 }
4989 }
4990 for ((parent_key, field), children) in buckets {
4991 if let Some(&idx) = idx_by_pk.get(&parent_key) {
4992 typed[idx].set_m2m_resolved_json(&field, children);
4993 }
4994 }
4995 // LEFT JOIN miss handling: walk every (parent, field) pair
4996 // we expected to populate and zero-init any slot that never
4997 // got a hit. Without this a parent with no matching M2M
4998 // children would leave its slot None — distinguishable from
4999 // "loaded, empty" only by callers checking the absence.
5000 for (parent_key, &idx) in idx_by_pk.iter() {
5001 for field in m2m_join_fields {
5002 let seg = field.split("__").next().unwrap_or(field.as_str());
5003 if !seen_children.contains_key(&(parent_key.clone(), seg.to_string())) {
5004 typed[idx].set_m2m_resolved_json(seg, Vec::new());
5005 }
5006 }
5007 }
5008 Ok(typed)
5009}
5010
5011fn dedup_decode_pg<T: Model + HydrateRelated>(
5012 raw_rows: &[sqlx::postgres::PgRow],
5013 fk_join_fields: &[String],
5014 m2m_join_fields: &[String],
5015) -> Result<Vec<T>, sqlx::Error>
5016where
5017 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
5018{
5019 // PK-agnostic dedup — see the SQLite variant.
5020 let parent_pk_col = crate::migrate::ModelMeta::for_::<T>()
5021 .fields
5022 .into_iter()
5023 .find(|c| c.primary_key)
5024 .ok_or_else(|| {
5025 sqlx::Error::Protocol(format!(
5026 "umbral::orm::join_related: model `{}` has no primary key, M2M JOIN \
5027 dedup requires one",
5028 T::NAME
5029 ))
5030 })?;
5031 let registered = crate::migrate::registered_models();
5032 let mut typed: Vec<T> = Vec::new();
5033 let mut idx_by_pk: HashMap<String, usize> = HashMap::new();
5034 let mut buckets: HashMap<(String, String), Vec<JsonValue>> = HashMap::new();
5035 let mut seen_children: HashMap<(String, String), std::collections::HashSet<String>> =
5036 HashMap::new();
5037 for row in raw_rows {
5038 let Ok(parent_json) = crate::orm::dynamic::decode_pg_to_json(row, &parent_pk_col) else {
5039 continue;
5040 };
5041 let parent_key = crate::orm::pk_key(&parent_json);
5042 if let std::collections::hash_map::Entry::Vacant(e) = idx_by_pk.entry(parent_key.clone()) {
5043 let mut t = <T as sqlx::FromRow<_>>::from_row(row)?;
5044 backend_pg::hydrate_joined_rels::<T>(&mut t, row, fk_join_fields)?;
5045 e.insert(typed.len());
5046 typed.push(t);
5047 }
5048 for m2m_field in m2m_join_fields {
5049 let m2m_seg = m2m_field.split("__").next().unwrap_or(m2m_field.as_str());
5050 let Some(rel) = T::M2M_RELATIONS.iter().find(|r| r.field_name == m2m_seg) else {
5051 continue;
5052 };
5053 let Some(child_meta) = registered.iter().find(|m| m.table == rel.target_table) else {
5054 continue;
5055 };
5056 let Some(child_json) =
5057 backend_pg::extract_m2m_child_json::<T>(row, m2m_field, child_meta)?
5058 else {
5059 continue;
5060 };
5061 let child_key = child_json
5062 .as_object()
5063 .and_then(|m| {
5064 let pk_col = child_meta.fields.iter().find(|c| c.primary_key)?;
5065 m.get(&pk_col.name).map(crate::orm::pk_key)
5066 })
5067 .unwrap_or_default();
5068 let key = (parent_key.clone(), m2m_seg.to_string());
5069 let seen = seen_children.entry(key.clone()).or_default();
5070 if seen.insert(child_key) {
5071 buckets.entry(key).or_default().push(child_json);
5072 }
5073 }
5074 }
5075 for ((parent_key, field), children) in buckets {
5076 if let Some(&idx) = idx_by_pk.get(&parent_key) {
5077 typed[idx].set_m2m_resolved_json(&field, children);
5078 }
5079 }
5080 // Same LEFT JOIN miss zero-init as the SQLite path.
5081 for (parent_key, &idx) in idx_by_pk.iter() {
5082 for field in m2m_join_fields {
5083 let seg = field.split("__").next().unwrap_or(field.as_str());
5084 if !seen_children.contains_key(&(parent_key.clone(), seg.to_string())) {
5085 typed[idx].set_m2m_resolved_json(seg, Vec::new());
5086 }
5087 }
5088 }
5089 Ok(typed)
5090}