1use std::collections::HashMap;
30
31use sea_query::{
32 Alias, Asterisk, Condition, Expr, Func, Order, PostgresQueryBuilder, Query, SqliteQueryBuilder,
33 Value as SeaValue,
34};
35use sea_query_binder::SqlxBinder;
36use sqlx::Row;
37
38use crate::db::{DbPool, pool_for_dispatched};
39use crate::migrate::{Column, ModelMeta};
40use crate::orm::SqlType;
41use crate::orm::write::{WriteError, json_to_sea_value, null_for};
42
43fn resolve_pool_dyn(meta: &crate::migrate::ModelMeta, op: crate::db::RouteOp) -> crate::db::DbPool {
46 let ctx = crate::db::route_context();
47 let r = crate::db::router::router();
48 let alias = match op {
49 crate::db::RouteOp::Read => r.db_for_read(meta, &ctx),
50 crate::db::RouteOp::Write => r.db_for_write(meta, &ctx),
51 };
52 pool_for_dispatched(alias.as_str()).clone()
53}
54
55#[derive(Debug)]
78pub enum DynError {
79 Write(WriteError),
84 Sqlx(sqlx::Error),
88}
89
90impl std::fmt::Display for DynError {
91 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
92 match self {
93 Self::Write(e) => write!(f, "{e}"),
94 Self::Sqlx(e) => write!(f, "{e}"),
95 }
96 }
97}
98
99impl std::error::Error for DynError {
100 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
101 match self {
102 Self::Write(e) => Some(e),
103 Self::Sqlx(e) => Some(e),
104 }
105 }
106}
107
108impl From<sqlx::Error> for DynError {
109 fn from(e: sqlx::Error) -> Self {
110 Self::Sqlx(e)
111 }
112}
113
114impl From<WriteError> for DynError {
115 fn from(e: WriteError) -> Self {
116 Self::Write(e)
117 }
118}
119
120pub struct DynQuerySet<'a> {
127 meta: &'a ModelMeta,
128 where_clauses: Vec<Condition>,
133 order: Vec<(String, bool)>,
134 limit: Option<u64>,
135 offset: Option<u64>,
136 select_cols: Vec<String>,
137 with_deleted: bool,
138 only_deleted: bool,
139 hard_delete: bool,
140 select_related: Vec<String>,
148 allow_privileged: Vec<String>,
154}
155
156pub fn typed_json_value(col: &Column, value: &str) -> Option<serde_json::Value> {
179 use serde_json::Value as J;
180 match crate::migrate::fk_effective_type(col) {
181 SqlType::SmallInt | SqlType::Integer | SqlType::BigInt | SqlType::ForeignKey => {
182 value.parse::<i64>().ok().map(J::from)
183 }
184 SqlType::Real | SqlType::Double => value.parse::<f64>().ok().map(J::from),
185 SqlType::Boolean => Some(J::Bool(matches!(value, "true" | "on" | "1"))),
186 SqlType::Uuid => uuid::Uuid::parse_str(value)
189 .ok()
190 .map(|u| J::String(u.to_string())),
191 _ => Some(J::String(value.to_string())),
192 }
193}
194
195pub fn never_matches() -> Condition {
201 Condition::all().add(Expr::val(1).eq(0))
202}
203
204pub fn typed_eq_condition(meta: &ModelMeta, col: &str, value: &str) -> Option<Condition> {
219 let meta_col = meta.fields.iter().find(|c| c.name == col)?;
220 let expr = Expr::col(Alias::new(col));
221 let predicate = match crate::migrate::fk_effective_type(meta_col) {
224 SqlType::SmallInt | SqlType::Integer => value.parse::<i32>().ok().map(|v| expr.eq(v)),
225 SqlType::BigInt | SqlType::ForeignKey => value.parse::<i64>().ok().map(|v| expr.eq(v)),
226 SqlType::Real | SqlType::Double => value.parse::<f64>().ok().map(|v| expr.eq(v)),
227 SqlType::Boolean => {
228 let v = matches!(value, "true" | "on" | "1");
229 Some(expr.eq(v))
230 }
231 SqlType::Uuid => uuid::Uuid::parse_str(value).ok().map(|u| expr.eq(u)),
234 _ => Some(expr.eq(value.to_string())),
235 };
236 predicate.map(|p| Condition::all().add(p))
237}
238
239impl<'a> DynQuerySet<'a> {
240 pub fn for_meta(meta: &'a ModelMeta) -> Self {
244 let select_cols = meta.fields.iter().map(|c| c.name.clone()).collect();
245 Self {
246 meta,
247 where_clauses: Vec::new(),
248 order: Vec::new(),
249 limit: None,
250 offset: None,
251 select_cols,
252 with_deleted: false,
253 only_deleted: false,
254 hard_delete: false,
255 select_related: Vec::new(),
256 allow_privileged: Vec::new(),
257 }
258 }
259
260 pub fn allow_privileged(mut self, cols: &[&str]) -> Self {
278 self.allow_privileged
279 .extend(cols.iter().map(|c| c.to_string()));
280 self
281 }
282
283 pub fn with_deleted(mut self) -> Self {
286 self.with_deleted = true;
287 self
288 }
289
290 pub fn only_deleted(mut self) -> Self {
293 self.only_deleted = true;
294 self
295 }
296
297 pub fn hard_delete(mut self) -> Self {
299 self.hard_delete = true;
300 self
301 }
302
303 fn where_clauses_with(&self, trashed: Option<bool>) -> Vec<Condition> {
317 let mut clauses = self.where_clauses.clone();
318 if self.meta.soft_delete {
319 let want_trashed = match trashed {
320 Some(t) => Some(t),
321 None if self.only_deleted => Some(true),
322 None if !self.with_deleted => Some(false),
323 None => None,
324 };
325 match want_trashed {
326 Some(true) => clauses
327 .push(Condition::all().add(Expr::col(Alias::new("deleted_at")).is_not_null())),
328 Some(false) => clauses
329 .push(Condition::all().add(Expr::col(Alias::new("deleted_at")).is_null())),
330 None => {}
331 }
332 }
333 clauses
334 }
335
336 fn effective_where_clauses(&self) -> Vec<Condition> {
337 self.where_clauses_with(None)
338 }
339
340 fn live_where_clauses(&self) -> Vec<Condition> {
341 self.where_clauses_with(Some(false))
342 }
343
344 fn trashed_where_clauses(&self) -> Vec<Condition> {
348 self.where_clauses_with(Some(true))
349 }
350
351 pub fn select_cols(mut self, cols: &[String]) -> Self {
355 let valid: Vec<String> = cols
356 .iter()
357 .filter(|n| self.meta.fields.iter().any(|c| &c.name == *n))
358 .cloned()
359 .collect();
360 if !valid.is_empty() {
361 self.select_cols = valid;
362 }
363 self
364 }
365
366 pub fn select_related_dyn(mut self, fields: &[String]) -> Self {
393 for name in fields {
394 let canonical = normalize_sr_token(name);
395 if validate_sr_chain(self.meta, &canonical).is_none() {
396 continue;
397 }
398 if !self.select_related.iter().any(|n| n == &canonical) {
399 self.select_related.push(canonical);
400 }
401 }
402 self
403 }
404
405 #[doc(hidden)]
408 pub fn select_related_fields(&self) -> &[String] {
409 &self.select_related
410 }
411
412 pub fn search(mut self, fields: &[String], term: &str) -> Self {
436 let term = term.trim();
437 if term.is_empty() {
438 return self;
439 }
440
441 let restricted = !fields.is_empty();
442 let as_int = term.parse::<i64>().ok();
443 let as_float = term.parse::<f64>().ok();
444 let as_bool = match term.to_ascii_lowercase().as_str() {
445 "true" => Some(true),
446 "false" => Some(false),
447 _ => None,
448 };
449 let like_pat = format!("%{}%", crate::orm::escape_like_literal(term)).to_uppercase();
452
453 let mut cond = Condition::any();
454 let mut added = 0;
455 for col in &self.meta.fields {
456 if restricted && !fields.iter().any(|f| f == &col.name) {
457 continue;
458 }
459 let predicate: Option<sea_query::SimpleExpr> = match col.ty {
460 SqlType::Text => Some(
461 Expr::expr(Func::upper(Expr::col(Alias::new(&col.name))))
462 .like(sea_query::LikeExpr::new(like_pat.clone()).escape('\\')),
463 ),
464 SqlType::SmallInt | SqlType::Integer | SqlType::BigInt | SqlType::ForeignKey => {
465 as_int.map(|n| Expr::col(Alias::new(&col.name)).eq(n))
466 }
467 SqlType::Real | SqlType::Double => {
468 as_float.map(|n| Expr::col(Alias::new(&col.name)).eq(n))
469 }
470 SqlType::Boolean => as_bool.map(|b| Expr::col(Alias::new(&col.name)).eq(b)),
471 _ => None,
472 };
473 if let Some(p) = predicate {
474 cond = cond.add(p);
475 added += 1;
476 }
477 }
478 if added > 0 {
479 self.where_clauses.push(cond);
480 }
481 self
482 }
483
484 pub fn filter_condition(mut self, cond: sea_query::Condition) -> Self {
490 self.where_clauses.push(cond);
491 self
492 }
493
494 pub fn filter_in_i64(mut self, col: &str, vals: &[i64]) -> Self {
497 if vals.is_empty() || !self.meta.fields.iter().any(|c| c.name == col) {
498 return self;
499 }
500 let cond = Condition::all().add(Expr::col(Alias::new(col)).is_in(vals.iter().copied()));
501 self.where_clauses.push(cond);
502 self
503 }
504
505 pub fn filter_m2m_contains_any(mut self, field_name: &str, child_ids: &[String]) -> Self {
529 if child_ids.is_empty() {
530 return self;
531 }
532 let Some(rel) = self
533 .meta
534 .m2m_relations
535 .iter()
536 .find(|r| r.field_name == field_name)
537 else {
538 return self;
539 };
540 let Some(pk_col) = self.meta.pk_column() else {
541 return self;
542 };
543 let target_pk_ty = crate::migrate::pk_meta_for_table(&rel.target_table)
554 .map(|(_, ty)| ty)
555 .unwrap_or(SqlType::BigInt);
556 let junction_table = format!("{}_{}", self.meta.table, rel.field_name);
557 let child_id_expr = Expr::col(Alias::new("child_id"));
558 let in_clause: sea_query::SimpleExpr = match target_pk_ty {
559 SqlType::Text | SqlType::Uuid => {
560 let bound: Vec<String> = child_ids
564 .iter()
565 .filter_map(|s| {
566 let s = s.trim();
567 if s.is_empty() {
568 None
569 } else {
570 Some(s.to_string())
571 }
572 })
573 .collect();
574 if bound.is_empty() {
575 return self;
576 }
577 child_id_expr.is_in(bound)
578 }
579 _ => {
580 let parsed: Vec<i64> = child_ids.iter().filter_map(|s| s.parse().ok()).collect();
584 if parsed.is_empty() {
585 return self;
586 }
587 child_id_expr.is_in(parsed)
588 }
589 };
590 let subq = Query::select()
591 .column(Alias::new("parent_id"))
592 .from(crate::db::router::schema_qualified_table(&junction_table))
593 .and_where(in_clause)
594 .to_owned();
595 let cond =
596 Condition::all().add(Expr::col(Alias::new(pk_col.name.clone())).in_subquery(subq));
597 self.where_clauses.push(cond);
598 self
599 }
600
601 pub fn filter_in_strings(mut self, col: &str, vals: &[String]) -> Self {
613 let Some(meta_col) = self.meta.fields.iter().find(|c| c.name == col) else {
614 return self;
615 };
616 if vals.is_empty() {
617 return self;
618 }
619 let expr = Expr::col(Alias::new(col));
620 let cond = match crate::migrate::fk_effective_type(meta_col) {
627 SqlType::SmallInt | SqlType::Integer => {
628 let parsed: Vec<i32> = vals.iter().filter_map(|s| s.parse().ok()).collect();
629 if parsed.is_empty() {
630 return self;
631 }
632 Condition::all().add(expr.is_in(parsed))
633 }
634 SqlType::BigInt | SqlType::ForeignKey => {
635 let parsed: Vec<i64> = vals.iter().filter_map(|s| s.parse().ok()).collect();
636 if parsed.is_empty() {
637 return self;
638 }
639 Condition::all().add(expr.is_in(parsed))
640 }
641 SqlType::Real | SqlType::Double => {
642 let parsed: Vec<f64> = vals.iter().filter_map(|s| s.parse().ok()).collect();
643 if parsed.is_empty() {
644 return self;
645 }
646 Condition::all().add(expr.is_in(parsed))
647 }
648 SqlType::Boolean => {
649 let parsed: Vec<bool> = vals
650 .iter()
651 .map(|s| matches!(s.as_str(), "true" | "on" | "1"))
652 .collect();
653 Condition::all().add(expr.is_in(parsed))
654 }
655 SqlType::Uuid => {
660 let parsed: Vec<uuid::Uuid> = vals
661 .iter()
662 .filter_map(|s| uuid::Uuid::parse_str(s).ok())
663 .collect();
664 if parsed.is_empty() {
665 return self;
666 }
667 Condition::all().add(expr.is_in(parsed))
668 }
669 _ => Condition::all().add(expr.is_in(vals.iter().map(|s| s.to_string()))),
670 };
671 self.where_clauses.push(cond);
672 self
673 }
674
675 pub fn filter_eq_string(mut self, col: &str, value: &str) -> Self {
679 match typed_eq_condition(self.meta, col, value) {
695 Some(cond) => self.where_clauses.push(cond),
696 None => self.where_clauses.push(never_matches()),
697 }
698 self
699 }
700
701 pub fn order_by_col(mut self, col: &str, descending: bool) -> Self {
704 if self.meta.fields.iter().any(|c| c.name == col) {
705 self.order.push((col.to_string(), descending));
706 }
707 self
708 }
709
710 pub fn limit(mut self, n: u64) -> Self {
712 self.limit = Some(n);
713 self
714 }
715
716 pub fn offset(mut self, n: u64) -> Self {
718 self.offset = Some(n);
719 self
720 }
721
722 pub async fn exists(self) -> Result<bool, DynError> {
734 let mut q = Query::select();
735 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
736 q.expr(Expr::val(1));
737 let where_clauses = self.effective_where_clauses();
738 for cond in &where_clauses {
739 q.cond_where(cond.clone());
740 }
741 q.limit(1);
742
743 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Read) {
744 DbPool::Sqlite(pool) => {
745 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
746 let row = sqlx::query_with(&sql, values).fetch_optional(&pool).await?;
747 Ok(row.is_some())
748 }
749 DbPool::Postgres(pool) => {
750 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
751 let row = sqlx::query_with(&sql, values).fetch_optional(&pool).await?;
752 Ok(row.is_some())
753 }
754 }
755 }
756
757 pub async fn count(self) -> Result<i64, DynError> {
758 let mut q = Query::select();
759 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
760 q.expr(Func::count(Expr::col(Asterisk)));
761 let where_clauses = self.effective_where_clauses();
762 for cond in &where_clauses {
763 q.cond_where(cond.clone());
764 }
765
766 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Read) {
767 DbPool::Sqlite(pool) => {
768 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
769 let row = sqlx::query_with(&sql, values).fetch_one(&pool).await?;
770 Ok(row.try_get::<i64, _>(0)?)
771 }
772 DbPool::Postgres(pool) => {
773 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
774 let row = sqlx::query_with(&sql, values).fetch_one(&pool).await?;
775 Ok(row.try_get::<i64, _>(0)?)
776 }
777 }
778 }
779
780 pub async fn fetch_distinct_strings(self, col: &str) -> Result<Vec<String>, DynError> {
785 let Some(col_meta) = self.meta.fields.iter().find(|c| c.name == col) else {
786 return Ok(Vec::new());
787 };
788 let mut q = Query::select();
789 q.distinct();
790 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
791 q.column(Alias::new(col));
792 let where_clauses = self.effective_where_clauses();
793 for cond in &where_clauses {
794 q.cond_where(cond.clone());
795 }
796 if let Some(n) = self.limit {
797 q.limit(n);
798 }
799
800 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Read) {
801 DbPool::Sqlite(pool) => {
802 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
803 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
804 let mut out = Vec::with_capacity(rows.len());
805 for row in rows {
806 out.push(decode_to_string(&row, col_meta)?);
807 }
808 Ok(out)
809 }
810 DbPool::Postgres(pool) => {
811 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
812 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
813 let mut out = Vec::with_capacity(rows.len());
814 for row in rows {
815 out.push(decode_pg_to_string(&row, col_meta)?);
816 }
817 Ok(out)
818 }
819 }
820 }
821
822 fn ensure_writable(&self) -> Result<(), crate::orm::write::WriteError> {
837 if self.meta.view.is_some() {
838 return Err(crate::orm::write::WriteError::ReadOnlyView {
839 table: self.meta.table.clone(),
840 });
841 }
842 Ok(())
843 }
844
845 pub async fn delete(self) -> Result<u64, DynError> {
846 self.ensure_writable()?;
847 if self.meta.soft_delete && !self.hard_delete {
848 return self.soft_delete_update().await;
849 }
850 let where_clauses = self.effective_where_clauses();
851 let audit_before = if self.meta.audited {
854 audit_snapshot(self.meta, &where_clauses).await
855 } else {
856 Vec::new()
857 };
858 let parent_pks: Vec<serde_json::Value> = match self.meta.pk_column() {
862 Some(pk_col) => collect_parent_pks(&self.meta, pk_col, &where_clauses)
863 .await
864 .unwrap_or_default(),
865 None => Vec::new(),
866 };
867
868 let mut q = Query::delete();
869 q.from_table(crate::db::router::schema_qualified_table(&self.meta.table));
870 for cond in &where_clauses {
871 q.cond_where(cond.clone());
872 }
873
874 let rows_affected = match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
875 DbPool::Sqlite(pool) => {
876 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
877 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
878 res.rows_affected()
879 }
880 DbPool::Postgres(pool) => {
881 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
882 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
883 res.rows_affected()
884 }
885 };
886
887 crate::signals::emit_bulk_post_delete_by_table(&self.meta.table, parent_pks).await;
892 if self.meta.audited {
893 let pairs = audit_pairs(self.meta, audit_before, Vec::new());
894 crate::orm::audit::record_many(self.meta, crate::orm::audit::DELETE, pairs).await;
895 }
896 Ok(rows_affected)
897 }
898
899 async fn soft_delete_update(self) -> Result<u64, DynError> {
900 let where_clauses = self.live_where_clauses();
901 let audit_before = if self.meta.audited {
905 audit_snapshot(self.meta, &where_clauses).await
906 } else {
907 Vec::new()
908 };
909 let parent_pks: Vec<serde_json::Value> = match self.meta.pk_column() {
910 Some(pk_col) => collect_parent_pks(self.meta, pk_col, &where_clauses)
911 .await
912 .unwrap_or_default(),
913 None => Vec::new(),
914 };
915
916 let at = chrono::Utc::now();
920 let meta = self.meta.clone();
921 let clauses = where_clauses.clone();
922
923 let rows_affected = crate::db::transaction(move |tx| {
926 Box::pin(async move {
927 if let Some(pk) = meta.pk_column() {
930 let mut sel = Query::select();
931 sel.column(Alias::new(&pk.name))
932 .from(crate::db::router::schema_qualified_table(&meta.table));
933 for cond in &clauses {
934 sel.cond_where(cond.clone());
935 }
936 let mut c = crate::orm::soft_delete_cascade::CascadeConn::from_tx(tx);
937 crate::orm::soft_delete_cascade::cascade_soft_delete(&mut c, &meta, sel, at)
938 .await?;
939 }
940
941 let mut q = Query::update();
942 q.table(crate::db::router::schema_qualified_table(&meta.table));
943 q.value(
944 Alias::new("deleted_at"),
945 sea_query::Value::ChronoDateTimeUtc(Some(Box::new(at))),
946 );
947 for cond in &clauses {
948 q.cond_where(cond.clone());
949 }
950 let affected = match tx.backend_name() {
951 "sqlite" => {
952 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
953 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
954 sqlx::query_with(&sql, values)
955 .execute(&mut **inner)
956 .await?
957 .rows_affected()
958 }
959 _ => {
960 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
961 let inner = tx.as_pg_mut().expect("postgres backend_name");
962 sqlx::query_with(&sql, values)
963 .execute(&mut **inner)
964 .await?
965 .rows_affected()
966 }
967 };
968 Ok::<u64, sqlx::Error>(affected)
969 })
970 })
971 .await?;
972
973 crate::signals::emit_bulk_post_delete_by_table(&self.meta.table, parent_pks).await;
974 if self.meta.audited {
975 let pairs = audit_pairs(self.meta, audit_before, Vec::new());
976 crate::orm::audit::record_many(self.meta, crate::orm::audit::DELETE, pairs).await;
977 }
978 Ok(rows_affected)
979 }
980
981 pub async fn restore(self) -> Result<u64, DynError> {
992 if !self.meta.soft_delete {
993 return Ok(0);
994 }
995 let where_clauses = self.trashed_where_clauses();
999
1000 let parent_pks: Vec<serde_json::Value> = match self.meta.pk_column() {
1001 Some(pk_col) => collect_parent_pks(self.meta, pk_col, &where_clauses)
1002 .await
1003 .unwrap_or_default(),
1004 None => Vec::new(),
1005 };
1006
1007 let meta = self.meta.clone();
1008 let clauses = where_clauses.clone();
1009
1010 let rows_affected = crate::db::transaction(move |tx| {
1011 Box::pin(async move {
1012 let ats = {
1019 let mut c = crate::orm::soft_delete_cascade::CascadeConn::from_tx(tx);
1020 crate::orm::soft_delete_cascade::deleted_at_values(&mut c, &meta, &clauses)
1021 .await?
1022 };
1023 for at in ats {
1024 if let Some(sel) =
1025 crate::orm::soft_delete_cascade::selector_at(&meta, &clauses, at)
1026 {
1027 let mut c = crate::orm::soft_delete_cascade::CascadeConn::from_tx(tx);
1028 crate::orm::soft_delete_cascade::cascade_restore(&mut c, &meta, sel, at)
1029 .await?;
1030 }
1031 }
1032
1033 let mut q = Query::update();
1034 q.table(crate::db::router::schema_qualified_table(&meta.table));
1035 q.value(
1036 Alias::new("deleted_at"),
1037 sea_query::Value::ChronoDateTimeUtc(None),
1038 );
1039 for cond in &clauses {
1040 q.cond_where(cond.clone());
1041 }
1042 let affected = match tx.backend_name() {
1043 "sqlite" => {
1044 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1045 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1046 sqlx::query_with(&sql, values)
1047 .execute(&mut **inner)
1048 .await?
1049 .rows_affected()
1050 }
1051 _ => {
1052 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1053 let inner = tx.as_pg_mut().expect("postgres backend_name");
1054 sqlx::query_with(&sql, values)
1055 .execute(&mut **inner)
1056 .await?
1057 .rows_affected()
1058 }
1059 };
1060 Ok::<u64, sqlx::Error>(affected)
1061 })
1062 })
1063 .await?;
1064
1065 crate::signals::emit_bulk_post_save_by_table(&self.meta.table, parent_pks, false).await;
1069 Ok(rows_affected)
1070 }
1071
1072 pub async fn update_one(self, col: &str, value: &str) -> Result<u64, DynError> {
1077 self.ensure_writable()?;
1078 let Some(col_meta) = self.meta.fields.iter().find(|c| c.name == col) else {
1079 return Ok(0);
1080 };
1081 let sea_value = match form_str_to_sea_value(col_meta, value) {
1082 Ok(v) => v,
1083 Err(e) => {
1085 return Err(DynError::Write(WriteError::Validator {
1086 field: col_meta.name.clone(),
1087 message: e.to_string(),
1088 }));
1089 }
1090 };
1091
1092 let mut q = Query::update();
1093 q.table(crate::db::router::schema_qualified_table(&self.meta.table));
1094 q.value(Alias::new(col), sea_value);
1095 let where_clauses = self.effective_where_clauses();
1096 for cond in &where_clauses {
1097 q.cond_where(cond.clone());
1098 }
1099
1100 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
1101 DbPool::Sqlite(pool) => {
1102 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1103 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
1104 Ok(res.rows_affected())
1105 }
1106 DbPool::Postgres(pool) => {
1107 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1108 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
1109 Ok(res.rows_affected())
1110 }
1111 }
1112 }
1113
1114 pub async fn update_form(
1121 self,
1122 form: &HashMap<String, String>,
1123 skip: &[String],
1124 ) -> Result<u64, DynError> {
1125 self.ensure_writable()?;
1126 let Some(q) = self.build_update_form_query(form, skip)? else {
1127 return Ok(0);
1128 };
1129
1130 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
1131 DbPool::Sqlite(pool) => {
1132 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1133 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
1134 Ok(res.rows_affected())
1135 }
1136 DbPool::Postgres(pool) => {
1137 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1138 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
1139 Ok(res.rows_affected())
1140 }
1141 }
1142 }
1143
1144 fn build_update_form_query(
1152 &self,
1153 form: &HashMap<String, String>,
1154 skip: &[String],
1155 ) -> Result<Option<sea_query::UpdateStatement>, DynError> {
1156 let mut q = Query::update();
1157 q.table(crate::db::router::schema_qualified_table(&self.meta.table));
1158 let mut any = false;
1159 for col in &self.meta.fields {
1160 if col.primary_key || skip.iter().any(|s| s == &col.name) {
1161 continue;
1162 }
1163 if is_unauthorized_privileged(col, &self.allow_privileged) {
1167 continue;
1168 }
1169 if col.auto_user {
1181 q.value(
1182 Alias::new(&col.name),
1183 crate::orm::write::user_for_column(col.ty),
1184 );
1185 any = true;
1186 continue;
1187 }
1188 if col.auto_now {
1189 q.value(
1190 Alias::new(&col.name),
1191 crate::orm::write::now_for_column(col.ty),
1192 );
1193 any = true;
1194 continue;
1195 }
1196 let Some(raw) = form.get(&col.name) else {
1197 continue;
1198 };
1199 let sea_value = match form_str_to_sea_value(col, raw) {
1200 Ok(v) => v,
1201 Err(e) => {
1207 return Err(DynError::Write(WriteError::Validator {
1208 field: col.name.clone(),
1209 message: e.to_string(),
1210 }));
1211 }
1212 };
1213 q.value(Alias::new(&col.name), sea_value);
1214 any = true;
1215 }
1216 if !any {
1217 return Ok(None);
1218 }
1219 let where_clauses = self.effective_where_clauses();
1220 for cond in &where_clauses {
1221 q.cond_where(cond.clone());
1222 }
1223 Ok(Some(q))
1224 }
1225
1226 pub async fn update_form_in_tx(
1234 self,
1235 tx: &mut crate::db::Transaction,
1236 form: &HashMap<String, String>,
1237 skip: &[String],
1238 ) -> Result<u64, DynError> {
1239 self.ensure_writable()?;
1240 let Some(q) = self.build_update_form_query(form, skip)? else {
1241 return Ok(0);
1242 };
1243
1244 match tx.backend_name() {
1245 "sqlite" => {
1246 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1247 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1248 let res = sqlx::query_with(&sql, values).execute(&mut **inner).await?;
1249 Ok(res.rows_affected())
1250 }
1251 _ => {
1252 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1253 let inner = tx.as_pg_mut().expect("postgres backend_name");
1254 let res = sqlx::query_with(&sql, values).execute(&mut **inner).await?;
1255 Ok(res.rows_affected())
1256 }
1257 }
1258 }
1259
1260 pub async fn insert_form(
1266 self,
1267 form: &HashMap<String, String>,
1268 skip: &[String],
1269 ) -> Result<i64, DynError> {
1270 self.ensure_writable()?;
1271 let Some(mut q) = self.build_insert_form_query(form, skip)? else {
1272 return Ok(0);
1273 };
1274
1275 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
1276 DbPool::Sqlite(pool) => {
1277 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
1278 let res = sqlx::query_with(&sql, vals).execute(&pool).await?;
1279 Ok(res.last_insert_rowid())
1280 }
1281 DbPool::Postgres(pool) => {
1282 let pk_name = self
1288 .meta
1289 .fields
1290 .iter()
1291 .find(|c| c.primary_key)
1292 .map(|c| c.name.clone());
1293 if let Some(pk) = pk_name {
1294 q.returning_col(Alias::new(&pk));
1295 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1296 let row = sqlx::query_with(&sql, vals).fetch_one(&pool).await?;
1297 Ok(row.try_get::<i64, _>(pk.as_str()).unwrap_or(0))
1298 } else {
1299 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1300 let _ = sqlx::query_with(&sql, vals).execute(&pool).await?;
1301 Ok(0)
1302 }
1303 }
1304 }
1305 }
1306
1307 fn build_insert_form_query(
1316 &self,
1317 form: &HashMap<String, String>,
1318 skip: &[String],
1319 ) -> Result<Option<sea_query::InsertStatement>, DynError> {
1320 let mut cols: Vec<&str> = Vec::new();
1321 let mut values: Vec<SeaValue> = Vec::new();
1322 for col in &self.meta.fields {
1323 if skip.iter().any(|s| s == &col.name) {
1324 continue;
1325 }
1326 if is_unauthorized_privileged(col, &self.allow_privileged) {
1331 continue;
1332 }
1333 if col.primary_key
1336 && matches!(
1337 col.ty,
1338 SqlType::Integer | SqlType::BigInt | SqlType::SmallInt
1339 )
1340 && form.get(&col.name).is_none_or(|v| v.is_empty())
1341 {
1342 continue;
1343 }
1344 if col.auto_user_add || col.auto_user {
1354 cols.push(&col.name);
1355 values.push(crate::orm::write::user_for_column(col.ty));
1356 continue;
1357 }
1358 if (col.auto_now_add || col.auto_now)
1359 && form.get(&col.name).is_none_or(|v| v.is_empty())
1360 {
1361 cols.push(&col.name);
1362 values.push(crate::orm::write::now_for_column(col.ty));
1363 continue;
1364 }
1365 let raw = form.get(&col.name).map(|s| s.as_str()).unwrap_or("");
1366 let sea_value = match form_str_to_sea_value(col, raw) {
1367 Ok(v) => v,
1368 Err(e) => {
1371 return Err(DynError::Write(WriteError::Validator {
1372 field: col.name.clone(),
1373 message: e.to_string(),
1374 }));
1375 }
1376 };
1377 cols.push(&col.name);
1378 values.push(sea_value);
1379 }
1380 if cols.is_empty() {
1381 return Ok(None);
1382 }
1383
1384 let mut q = Query::insert();
1385 q.into_table(crate::db::router::schema_qualified_table(&self.meta.table));
1386 q.columns(cols.iter().map(|c| Alias::new(*c)).collect::<Vec<_>>());
1387 let exprs: Vec<sea_query::SimpleExpr> = values.into_iter().map(Into::into).collect();
1388 q.values_panic(exprs);
1389 Ok(Some(q))
1390 }
1391
1392 pub async fn insert_form_in_tx(
1401 self,
1402 tx: &mut crate::db::Transaction,
1403 form: &HashMap<String, String>,
1404 skip: &[String],
1405 ) -> Result<i64, DynError> {
1406 self.ensure_writable()?;
1407 let Some(mut q) = self.build_insert_form_query(form, skip)? else {
1408 return Ok(0);
1409 };
1410
1411 match tx.backend_name() {
1412 "sqlite" => {
1413 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
1414 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1415 let res = sqlx::query_with(&sql, vals).execute(&mut **inner).await?;
1416 Ok(res.last_insert_rowid())
1417 }
1418 _ => {
1419 let pk_name = self
1423 .meta
1424 .fields
1425 .iter()
1426 .find(|c| c.primary_key)
1427 .map(|c| c.name.clone());
1428 let inner = tx.as_pg_mut().expect("postgres backend_name");
1429 if let Some(pk) = pk_name {
1430 q.returning_col(Alias::new(&pk));
1431 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1432 let row = sqlx::query_with(&sql, vals).fetch_one(&mut **inner).await?;
1433 Ok(row.try_get::<i64, _>(pk.as_str()).unwrap_or(0))
1434 } else {
1435 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1436 let _ = sqlx::query_with(&sql, vals).execute(&mut **inner).await?;
1437 Ok(0)
1438 }
1439 }
1440 }
1441 }
1442
1443 pub async fn fetch_as_strings(self) -> Result<Vec<HashMap<String, String>>, DynError> {
1448 let mut q = Query::select();
1449 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
1450 for c in &self.select_cols {
1451 q.column(Alias::new(c));
1452 }
1453 let where_clauses = self.effective_where_clauses();
1454 for cond in &where_clauses {
1455 q.cond_where(cond.clone());
1456 }
1457 for (col, descending) in &self.order {
1458 q.order_by(
1459 Alias::new(col),
1460 if *descending { Order::Desc } else { Order::Asc },
1461 );
1462 }
1463 if let Some(n) = self.limit {
1464 q.limit(n);
1465 }
1466 if let Some(n) = self.offset {
1467 q.offset(n);
1468 }
1469
1470 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Read) {
1471 DbPool::Sqlite(pool) => {
1472 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1473 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
1474 let mut out: Vec<HashMap<String, String>> = Vec::with_capacity(rows.len());
1475 for row in rows {
1476 let mut entry = HashMap::new();
1477 for col_name in &self.select_cols {
1478 if let Some(col_meta) =
1479 self.meta.fields.iter().find(|c| &c.name == col_name)
1480 {
1481 let v = decode_to_string(&row, col_meta)?;
1482 entry.insert(col_name.clone(), v);
1483 }
1484 }
1485 out.push(entry);
1486 }
1487 Ok(out)
1488 }
1489 DbPool::Postgres(pool) => {
1490 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1491 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
1492 let mut out: Vec<HashMap<String, String>> = Vec::with_capacity(rows.len());
1493 for row in rows {
1494 let mut entry = HashMap::new();
1495 for col_name in &self.select_cols {
1496 if let Some(col_meta) =
1497 self.meta.fields.iter().find(|c| &c.name == col_name)
1498 {
1499 let v = decode_pg_to_string(&row, col_meta)?;
1500 entry.insert(col_name.clone(), v);
1501 }
1502 }
1503 out.push(entry);
1504 }
1505 Ok(out)
1506 }
1507 }
1508 }
1509
1510 pub async fn fetch_as_json(
1516 self,
1517 ) -> Result<Vec<serde_json::Map<String, serde_json::Value>>, DynError> {
1518 let mut q = Query::select();
1519 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
1520 for c in &self.select_cols {
1521 q.column(Alias::new(c));
1522 }
1523 let where_clauses = self.effective_where_clauses();
1524 for cond in &where_clauses {
1525 q.cond_where(cond.clone());
1526 }
1527 for (col, descending) in &self.order {
1528 q.order_by(
1529 Alias::new(col),
1530 if *descending { Order::Desc } else { Order::Asc },
1531 );
1532 }
1533 if let Some(n) = self.limit {
1534 q.limit(n);
1535 }
1536 if let Some(n) = self.offset {
1537 q.offset(n);
1538 }
1539
1540 let pk_name = self
1541 .meta
1542 .pk_column()
1543 .map(|c| c.name.clone())
1544 .unwrap_or_default();
1545 let selected_cols: Vec<(&String, &Column)> = self
1546 .select_cols
1547 .iter()
1548 .filter_map(|col_name| {
1549 self.meta
1550 .fields
1551 .iter()
1552 .find(|c| &c.name == col_name)
1553 .map(|col| (col_name, col))
1554 })
1555 .collect();
1556 let mut out: Vec<serde_json::Map<String, serde_json::Value>> =
1557 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Read) {
1558 DbPool::Sqlite(pool) => {
1559 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1560 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
1561 let mut out: Vec<serde_json::Map<String, serde_json::Value>> =
1562 Vec::with_capacity(rows.len());
1563 for row in rows {
1564 let mut entry = serde_json::Map::new();
1565 for (col_name, col_meta) in &selected_cols {
1566 entry.insert((*col_name).clone(), decode_to_json(&row, col_meta)?);
1567 }
1568 out.push(entry);
1569 }
1570 out
1571 }
1572 DbPool::Postgres(pool) => {
1573 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1574 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
1575 let mut out: Vec<serde_json::Map<String, serde_json::Value>> =
1576 Vec::with_capacity(rows.len());
1577 for row in rows {
1578 let mut entry = serde_json::Map::new();
1579 for (col_name, col_meta) in &selected_cols {
1580 entry.insert((*col_name).clone(), decode_pg_to_json(&row, col_meta)?);
1581 }
1582 out.push(entry);
1583 }
1584 out
1585 }
1586 };
1587
1588 if !self.meta.m2m_relations.is_empty() && !out.is_empty() {
1598 hydrate_m2m_batched(&self.meta, &pk_name, &mut out).await?;
1599 }
1600
1601 if !self.select_related.is_empty() && !out.is_empty() {
1610 hydrate_select_related_into(&self.meta, &self.select_related, &mut out).await?;
1611 }
1612 Ok(out)
1613 }
1614
1615 pub async fn first_as_json(
1618 mut self,
1619 ) -> Result<Option<serde_json::Map<String, serde_json::Value>>, DynError> {
1620 self.limit = Some(1);
1621 let mut rows = self.fetch_as_json().await?;
1622 Ok(rows.pop())
1623 }
1624
1625 pub async fn fetch_one_json_in_tx(
1635 self,
1636 tx: &mut crate::db::Transaction,
1637 ) -> Result<Option<serde_json::Map<String, serde_json::Value>>, DynError> {
1638 let mut q = Query::select();
1639 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
1640 for c in &self.meta.fields {
1641 q.column(Alias::new(&c.name));
1642 }
1643 let where_clauses = self.effective_where_clauses();
1644 for cond in &where_clauses {
1645 q.cond_where(cond.clone());
1646 }
1647 q.limit(1);
1648
1649 let out = match tx.backend_name() {
1650 "sqlite" => {
1651 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1652 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1653 let row = sqlx::query_with(&sql, values)
1654 .fetch_optional(&mut **inner)
1655 .await?;
1656 match row {
1657 Some(row) => {
1658 let mut entry = serde_json::Map::new();
1659 for col in &self.meta.fields {
1660 entry.insert(col.name.clone(), decode_to_json(&row, col)?);
1661 }
1662 Some(entry)
1663 }
1664 None => None,
1665 }
1666 }
1667 _ => {
1668 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1669 let inner = tx.as_pg_mut().expect("postgres backend_name");
1670 let row = sqlx::query_with(&sql, values)
1671 .fetch_optional(&mut **inner)
1672 .await?;
1673 match row {
1674 Some(row) => {
1675 let mut entry = serde_json::Map::new();
1676 for col in &self.meta.fields {
1677 entry.insert(col.name.clone(), decode_pg_to_json(&row, col)?);
1678 }
1679 Some(entry)
1680 }
1681 None => None,
1682 }
1683 }
1684 };
1685 Ok(out)
1686 }
1687
1688 pub async fn insert_json(
1696 self,
1697 body: &serde_json::Map<String, serde_json::Value>,
1698 ) -> Result<serde_json::Map<String, serde_json::Value>, crate::orm::write::WriteError> {
1699 self.ensure_writable()?;
1700 use crate::orm::write::WriteError;
1701
1702 let body_owned: serde_json::Map<String, serde_json::Value>;
1705 let body: &serde_json::Map<String, serde_json::Value> =
1706 match normalise_insert_body(self.meta, body, &self.allow_privileged) {
1707 Some(owned) => {
1708 body_owned = owned;
1709 &body_owned
1710 }
1711 None => body,
1712 };
1713
1714 let validation_errors = crate::orm::validation::validate_on_create(self.meta, body).await;
1716 if !validation_errors.is_empty() {
1717 return Err(WriteError::Multiple {
1718 errors: validation_errors,
1719 });
1720 }
1721
1722 let InsertPlan {
1725 mut q,
1726 pk_name,
1727 pk_ty,
1728 } = build_insert_plan(self.meta, body)?;
1729
1730 crate::signals::emit_pre_save_by_table(
1736 &self.meta.table,
1737 serde_json::Value::Object(body.clone()),
1738 true,
1739 )
1740 .await;
1741
1742 let mut tx = match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
1753 DbPool::Sqlite(pool) => crate::db::begin_sqlite(&pool).await,
1754 DbPool::Postgres(pool) => crate::db::begin_pg(&pool).await,
1755 }?;
1756
1757 let mut out = match tx.backend_name() {
1758 "sqlite" => {
1759 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
1760 let res = {
1761 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1762 sqlx::query_with(&sql, vals)
1763 .execute(&mut **inner)
1764 .await
1765 .map_err(|e| classify_or_sqlx(e, body))?
1766 };
1767 let pk_pred = match pk_ty {
1771 SqlType::Integer | SqlType::BigInt | SqlType::SmallInt => {
1772 Expr::col(Alias::new(&pk_name)).eq(res.last_insert_rowid())
1773 }
1774 _ => {
1775 let supplied = body
1778 .get(&pk_name)
1779 .cloned()
1780 .unwrap_or(serde_json::Value::Null);
1781 let sea_value = crate::orm::write::json_to_sea_value(
1782 pk_ty, &supplied, false, &pk_name, None,
1783 )?;
1784 Expr::col(Alias::new(&pk_name)).eq(sea_value)
1785 }
1786 };
1787 let mut sel = Query::select();
1788 sel.from(crate::db::router::schema_qualified_table(&self.meta.table));
1789 for c in &self.meta.fields {
1790 sel.column(Alias::new(&c.name));
1791 }
1792 sel.cond_where(Condition::all().add(pk_pred));
1793 let (sel_sql, sel_vals) = sel.build_sqlx(SqliteQueryBuilder);
1794 let mut out = serde_json::Map::new();
1795 {
1796 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1797 let row = sqlx::query_with(&sel_sql, sel_vals)
1798 .fetch_one(&mut **inner)
1799 .await?;
1800 for col in &self.meta.fields {
1801 out.insert(col.name.clone(), decode_to_json(&row, col)?);
1802 }
1803 }
1804 out
1805 }
1806 _ => {
1807 q.returning_all();
1812 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1813 let mut out = serde_json::Map::new();
1814 {
1815 let inner = tx.as_pg_mut().expect("postgres backend_name");
1816 let row = sqlx::query_with(&sql, vals)
1817 .fetch_one(&mut **inner)
1818 .await
1819 .map_err(|e| classify_or_sqlx(e, body))?;
1820 for col in &self.meta.fields {
1821 out.insert(col.name.clone(), decode_pg_to_json(&row, col)?);
1822 }
1823 }
1824 out
1825 }
1826 };
1827
1828 let pk_value = out.get(&pk_name).cloned();
1835 write_m2m_junctions_in_tx(self.meta, pk_value.as_ref(), body, &mut tx).await?;
1836 hydrate_m2m_into_tx(self.meta, pk_value.as_ref(), &mut out, &mut tx).await?;
1839
1840 tx.commit().await?;
1841
1842 crate::signals::emit_post_save_by_table(
1845 &self.meta.table,
1846 serde_json::Value::Object(out.clone()),
1847 true,
1848 )
1849 .await;
1850 Box::pin(crate::orm::audit::record(
1855 self.meta,
1856 &crate::orm::audit::pk_of(self.meta, &out),
1857 crate::orm::audit::CREATE,
1858 None,
1859 Some(&out),
1860 ))
1861 .await;
1862 Ok(out)
1863 }
1864
1865 pub async fn insert_json_in_tx(
1889 self,
1890 body: &serde_json::Map<String, serde_json::Value>,
1891 tx: &mut crate::db::Transaction,
1892 ) -> Result<serde_json::Map<String, serde_json::Value>, crate::orm::write::WriteError> {
1893 self.ensure_writable()?;
1894 use crate::orm::write::WriteError;
1895
1896 let body_owned: serde_json::Map<String, serde_json::Value>;
1898 let body: &serde_json::Map<String, serde_json::Value> =
1899 match normalise_insert_body(self.meta, body, &self.allow_privileged) {
1900 Some(owned) => {
1901 body_owned = owned;
1902 &body_owned
1903 }
1904 None => body,
1905 };
1906
1907 let validation_errors =
1910 crate::orm::validation::validate_on_create_in_tx(self.meta, body, tx).await;
1911 if !validation_errors.is_empty() {
1912 return Err(WriteError::Multiple {
1913 errors: validation_errors,
1914 });
1915 }
1916
1917 let InsertPlan {
1919 mut q,
1920 pk_name,
1921 pk_ty,
1922 } = build_insert_plan(self.meta, body)?;
1923
1924 match tx.backend_name() {
1925 "sqlite" => {
1926 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
1927 let res = {
1928 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1929 sqlx::query_with(&sql, vals)
1930 .execute(&mut **inner)
1931 .await
1932 .map_err(|e| classify_or_sqlx(e, body))?
1933 };
1934 let pk_pred = match pk_ty {
1937 SqlType::Integer | SqlType::BigInt | SqlType::SmallInt => {
1938 Expr::col(Alias::new(&pk_name)).eq(res.last_insert_rowid())
1939 }
1940 _ => {
1941 let supplied = body
1942 .get(&pk_name)
1943 .cloned()
1944 .unwrap_or(serde_json::Value::Null);
1945 let sea_value = crate::orm::write::json_to_sea_value(
1946 pk_ty, &supplied, false, &pk_name, None,
1947 )?;
1948 Expr::col(Alias::new(&pk_name)).eq(sea_value)
1949 }
1950 };
1951 let mut sel = Query::select();
1952 sel.from(crate::db::router::schema_qualified_table(&self.meta.table));
1953 for c in &self.meta.fields {
1954 sel.column(Alias::new(&c.name));
1955 }
1956 sel.cond_where(Condition::all().add(pk_pred));
1957 let (sel_sql, sel_vals) = sel.build_sqlx(SqliteQueryBuilder);
1958 let mut out = serde_json::Map::new();
1959 {
1960 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1961 let row = sqlx::query_with(&sel_sql, sel_vals)
1962 .fetch_one(&mut **inner)
1963 .await?;
1964 for col in &self.meta.fields {
1965 out.insert(col.name.clone(), decode_to_json(&row, col)?);
1966 }
1967 }
1968 let pk_value = out.get(&pk_name).cloned();
1970 write_m2m_junctions_in_tx(self.meta, pk_value.as_ref(), body, tx).await?;
1971 hydrate_m2m_into_tx(self.meta, pk_value.as_ref(), &mut out, tx).await?;
1972 Ok(out)
1973 }
1974 _ => {
1975 q.returning_all();
1976 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1977 let mut out = serde_json::Map::new();
1978 {
1979 let inner = tx.as_pg_mut().expect("postgres backend_name");
1980 let row = sqlx::query_with(&sql, vals)
1981 .fetch_one(&mut **inner)
1982 .await
1983 .map_err(|e| classify_or_sqlx(e, body))?;
1984 for col in &self.meta.fields {
1985 out.insert(col.name.clone(), decode_pg_to_json(&row, col)?);
1986 }
1987 }
1988 let pk_value = out.get(&pk_name).cloned();
1989 write_m2m_junctions_in_tx(self.meta, pk_value.as_ref(), body, tx).await?;
1990 hydrate_m2m_into_tx(self.meta, pk_value.as_ref(), &mut out, tx).await?;
1991 Ok(out)
1992 }
1993 }
1994 }
1995
1996 pub async fn update_json_in_tx(
2007 self,
2008 body: &serde_json::Map<String, serde_json::Value>,
2009 tx: &mut crate::db::Transaction,
2010 ) -> Result<u64, crate::orm::write::WriteError> {
2011 self.ensure_writable()?;
2012 use crate::orm::write::WriteError;
2013
2014 let body_owned: serde_json::Map<String, serde_json::Value>;
2017 let body: &serde_json::Map<String, serde_json::Value> =
2018 match normalise_update_body(self.meta, body, &self.allow_privileged) {
2019 Some(owned) => {
2020 body_owned = owned;
2021 &body_owned
2022 }
2023 None => body,
2024 };
2025
2026 let validation_errors =
2030 crate::orm::validation::validate_on_update_in_tx(self.meta, body, tx).await;
2031 if !validation_errors.is_empty() {
2032 return Err(WriteError::Multiple {
2033 errors: validation_errors,
2034 });
2035 }
2036
2037 let mut q = Query::update();
2038 q.table(crate::db::router::schema_qualified_table(&self.meta.table));
2039 let mut any = false;
2040 for col in &self.meta.fields {
2041 if col.primary_key {
2042 continue;
2043 }
2044 if col.auto_user {
2049 q.value(
2050 Alias::new(&col.name),
2051 crate::orm::write::user_for_column(col.ty),
2052 );
2053 any = true;
2054 continue;
2055 }
2056 let Some(json) = body.get(&col.name) else {
2057 if col.auto_now {
2058 let now_value = crate::orm::write::now_for_column(col.ty);
2059 q.value(Alias::new(&col.name), now_value);
2060 any = true;
2061 }
2062 continue;
2063 };
2064 validate_numeric_bounds(col, json)?;
2065 if let (Some(fmt), Some(s)) = (col.text_format.as_deref(), json.as_str()) {
2066 if let Err(e) = crate::orm::validators::validate_text_format(fmt, s) {
2067 return Err(WriteError::Validator {
2068 field: col.name.clone(),
2069 message: e.to_string(),
2070 });
2071 }
2072 }
2073 let normalized_json = normalize_json_for_col(col, json);
2077 let json = normalized_json.as_ref().unwrap_or(json);
2078 let cleaned_json = crate::orm::cleaners::apply(&self.meta.table, &col.name, json)?;
2081 let json = cleaned_json.as_ref().unwrap_or(json);
2082 let sealed = crate::orm::write::seal_masked_json(col, json)?;
2085 let sea_value = crate::orm::write::json_to_sea_value(
2086 col.ty,
2087 sealed.as_ref().unwrap_or(json),
2088 col.nullable,
2089 &col.name,
2090 fk_target_pk_sql_type(col),
2091 )?;
2092 q.value(Alias::new(&col.name), sea_value);
2093 any = true;
2094 }
2095 let touches_m2m = self
2096 .meta
2097 .m2m_relations
2098 .iter()
2099 .any(|r| body.contains_key(&r.field_name));
2100 if !any && !touches_m2m {
2101 return Ok(0);
2102 }
2103 let where_clauses = self.effective_where_clauses();
2104 for cond in &where_clauses {
2105 q.cond_where(cond.clone());
2106 }
2107
2108 let parent_pks: Vec<serde_json::Value> = match self.meta.pk_column() {
2112 Some(pk_col) => collect_parent_pks_in_tx(self.meta, pk_col, &where_clauses, tx).await?,
2113 None => Vec::new(),
2114 };
2115
2116 if any {
2117 match tx.backend_name() {
2118 "sqlite" => {
2119 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
2120 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
2121 sqlx::query_with(&sql, values)
2122 .execute(&mut **inner)
2123 .await
2124 .map_err(|e| classify_or_sqlx(e, body))?;
2125 }
2126 _ => {
2127 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
2128 let inner = tx.as_pg_mut().expect("postgres backend_name");
2129 sqlx::query_with(&sql, values)
2130 .execute(&mut **inner)
2131 .await
2132 .map_err(|e| classify_or_sqlx(e, body))?;
2133 }
2134 }
2135 }
2136 for pk in &parent_pks {
2137 write_m2m_junctions_in_tx(self.meta, Some(pk), body, tx).await?;
2138 }
2139 Ok(parent_pks.len().max(if any { 1 } else { 0 }) as u64)
2140 }
2141
2142 pub async fn delete_in_tx(self, tx: &mut crate::db::Transaction) -> Result<u64, DynError> {
2150 self.ensure_writable()?;
2151 let soft = self.meta.soft_delete && !self.hard_delete;
2152 let where_clauses = if soft {
2153 self.live_where_clauses()
2154 } else {
2155 self.effective_where_clauses()
2156 };
2157
2158 let table = crate::db::router::schema_qualified_table(&self.meta.table);
2162 let build = |is_sqlite: bool| {
2163 if soft {
2164 let mut u = Query::update();
2165 u.table(table.clone());
2166 u.value(
2167 Alias::new("deleted_at"),
2168 sea_query::Value::ChronoDateTimeUtc(Some(Box::new(chrono::Utc::now()))),
2169 );
2170 for cond in &where_clauses {
2171 u.cond_where(cond.clone());
2172 }
2173 if is_sqlite {
2174 u.build_sqlx(SqliteQueryBuilder)
2175 } else {
2176 u.build_sqlx(PostgresQueryBuilder)
2177 }
2178 } else {
2179 let mut d = Query::delete();
2180 d.from_table(table.clone());
2181 for cond in &where_clauses {
2182 d.cond_where(cond.clone());
2183 }
2184 if is_sqlite {
2185 d.build_sqlx(SqliteQueryBuilder)
2186 } else {
2187 d.build_sqlx(PostgresQueryBuilder)
2188 }
2189 }
2190 };
2191
2192 let rows_affected = match tx.backend_name() {
2193 "sqlite" => {
2194 let (sql, values) = build(true);
2195 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
2196 sqlx::query_with(&sql, values)
2197 .execute(&mut **inner)
2198 .await?
2199 .rows_affected()
2200 }
2201 _ => {
2202 let (sql, values) = build(false);
2203 let inner = tx.as_pg_mut().expect("postgres backend_name");
2204 sqlx::query_with(&sql, values)
2205 .execute(&mut **inner)
2206 .await?
2207 .rows_affected()
2208 }
2209 };
2210 Ok(rows_affected)
2211 }
2212
2213 pub async fn update_json(
2217 self,
2218 body: &serde_json::Map<String, serde_json::Value>,
2219 ) -> Result<u64, crate::orm::write::WriteError> {
2220 self.ensure_writable()?;
2221 use crate::orm::write::WriteError;
2222
2223 let body_owned: serde_json::Map<String, serde_json::Value>;
2230 let body: &serde_json::Map<String, serde_json::Value> =
2231 match normalise_update_body(self.meta, body, &self.allow_privileged) {
2232 Some(owned) => {
2233 body_owned = owned;
2234 &body_owned
2235 }
2236 None => body,
2237 };
2238
2239 let audit_before = if self.meta.audited {
2247 audit_snapshot(self.meta, &self.live_where_clauses()).await
2248 } else {
2249 Vec::new()
2250 };
2251
2252 let validation_errors = crate::orm::validation::validate_on_update(&self.meta, body).await;
2253 if !validation_errors.is_empty() {
2254 return Err(WriteError::Multiple {
2255 errors: validation_errors,
2256 });
2257 }
2258
2259 let mut q = Query::update();
2260 q.table(crate::db::router::schema_qualified_table(&self.meta.table));
2261 let mut any = false;
2262 for col in &self.meta.fields {
2263 if col.primary_key {
2264 continue;
2265 }
2266 if col.auto_user {
2271 q.value(
2272 Alias::new(&col.name),
2273 crate::orm::write::user_for_column(col.ty),
2274 );
2275 any = true;
2276 continue;
2277 }
2278 let Some(json) = body.get(&col.name) else {
2279 if col.auto_now {
2284 let now_value = crate::orm::write::now_for_column(col.ty);
2285 q.value(Alias::new(&col.name), now_value);
2286 any = true;
2287 }
2288 continue;
2289 };
2290 validate_numeric_bounds(col, json)?;
2291 if let (Some(fmt), Some(s)) = (col.text_format.as_deref(), json.as_str()) {
2294 if let Err(e) = crate::orm::validators::validate_text_format(fmt, s) {
2295 return Err(WriteError::Validator {
2296 field: col.name.clone(),
2297 message: e.to_string(),
2298 });
2299 }
2300 }
2301 let normalized_json = normalize_json_for_col(col, json);
2305 let json = normalized_json.as_ref().unwrap_or(json);
2306 let cleaned_json = crate::orm::cleaners::apply(&self.meta.table, &col.name, json)?;
2309 let json = cleaned_json.as_ref().unwrap_or(json);
2310 let sealed = crate::orm::write::seal_masked_json(col, json)?;
2313 let sea_value = crate::orm::write::json_to_sea_value(
2314 col.ty,
2315 sealed.as_ref().unwrap_or(json),
2316 col.nullable,
2317 &col.name,
2318 fk_target_pk_sql_type(col),
2319 )?;
2320 q.value(Alias::new(&col.name), sea_value);
2321 any = true;
2322 }
2323 let touches_m2m = self
2328 .meta
2329 .m2m_relations
2330 .iter()
2331 .any(|r| body.contains_key(&r.field_name));
2332 if !any && !touches_m2m {
2333 return Ok(0);
2334 }
2335 let where_clauses = self.effective_where_clauses();
2336 for cond in &where_clauses {
2337 q.cond_where(cond.clone());
2338 }
2339
2340 let mut tx = match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
2345 DbPool::Sqlite(pool) => crate::db::begin_sqlite(&pool).await,
2346 DbPool::Postgres(pool) => crate::db::begin_pg(&pool).await,
2347 }?;
2348
2349 let parent_pks: Vec<serde_json::Value> = match self.meta.pk_column() {
2363 Some(pk_col) => {
2364 collect_parent_pks_in_tx(self.meta, pk_col, &where_clauses, &mut tx).await?
2365 }
2366 None => Vec::new(),
2367 };
2368
2369 let rows_affected = if any {
2374 match tx.backend_name() {
2375 "sqlite" => {
2376 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
2377 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
2378 sqlx::query_with(&sql, values)
2379 .execute(&mut **inner)
2380 .await
2381 .map_err(|e| classify_or_sqlx(e, body))?
2382 .rows_affected()
2383 }
2384 _ => {
2385 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
2386 let inner = tx.as_pg_mut().expect("postgres backend_name");
2387 sqlx::query_with(&sql, values)
2388 .execute(&mut **inner)
2389 .await
2390 .map_err(|e| classify_or_sqlx(e, body))?
2391 .rows_affected()
2392 }
2393 }
2394 } else {
2395 0
2396 };
2397
2398 for pk in &parent_pks {
2399 write_m2m_junctions_in_tx(self.meta, Some(pk), body, &mut tx).await?;
2400 }
2401
2402 tx.commit().await?;
2403
2404 crate::signals::emit_bulk_post_save_by_table(&self.meta.table, parent_pks.clone(), false)
2409 .await;
2410
2411 if self.meta.audited && !audit_before.is_empty() {
2416 let pks: Vec<serde_json::Value> = parent_pks.clone();
2417 let after = match crate::orm::audit::pk_in_condition(self.meta, &pks) {
2418 Some(cond) => audit_snapshot(self.meta, &[cond]).await,
2419 None => Vec::new(),
2420 };
2421 let pairs = audit_pairs(self.meta, audit_before, after);
2422 crate::orm::audit::record_many(self.meta, crate::orm::audit::UPDATE, pairs).await;
2423 }
2424
2425 if any {
2430 Ok(rows_affected)
2431 } else {
2432 Ok(parent_pks.len() as u64)
2433 }
2434 }
2435}
2436
2437pub fn decode_to_string(
2443 row: &sqlx::sqlite::SqliteRow,
2444 col: &Column,
2445) -> Result<String, sqlx::Error> {
2446 use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
2447 use serde_json::Value;
2448 use uuid::Uuid;
2449
2450 let name = col.name.as_str();
2451 if col.nullable {
2452 return Ok(match col.ty {
2453 SqlType::SmallInt | SqlType::Integer => row
2454 .try_get::<Option<i32>, _>(name)?
2455 .map_or(String::new(), |v| v.to_string()),
2456 SqlType::BigInt => row
2457 .try_get::<Option<i64>, _>(name)?
2458 .map_or(String::new(), |v| v.to_string()),
2459 SqlType::Real => row
2460 .try_get::<Option<f32>, _>(name)?
2461 .map_or(String::new(), |v| v.to_string()),
2462 SqlType::Double => row
2463 .try_get::<Option<f64>, _>(name)?
2464 .map_or(String::new(), |v| v.to_string()),
2465 SqlType::Boolean => row
2466 .try_get::<Option<bool>, _>(name)?
2467 .map_or(String::new(), |v| {
2468 if v { "true" } else { "false" }.to_string()
2469 }),
2470 SqlType::Text => row.try_get::<Option<String>, _>(name)?.unwrap_or_default(),
2471 SqlType::Date => row
2472 .try_get::<Option<NaiveDate>, _>(name)?
2473 .map_or(String::new(), |v| v.to_string()),
2474 SqlType::Time => row
2475 .try_get::<Option<NaiveTime>, _>(name)?
2476 .map_or(String::new(), |v| v.to_string()),
2477 SqlType::Timestamptz => row
2478 .try_get::<Option<DateTime<Utc>>, _>(name)?
2479 .map_or(String::new(), |v| v.to_rfc3339()),
2480 SqlType::Uuid => row
2481 .try_get::<Option<Uuid>, _>(name)?
2482 .map_or(String::new(), |v| v.to_string()),
2483 SqlType::Json => row
2484 .try_get::<Option<Value>, _>(name)?
2485 .map_or(String::new(), |v| v.to_string()),
2486 SqlType::Array(_) => panic_array_unsupported(&col.name),
2487 SqlType::Inet
2488 | SqlType::Cidr
2489 | SqlType::MacAddr
2490 | SqlType::Xml
2491 | SqlType::Ltree
2492 | SqlType::Bit
2493 | SqlType::FullText => panic_pg_only_unsupported(&col.name),
2494 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2498 Some(SqlType::Text) => row.try_get::<Option<String>, _>(name)?.unwrap_or_default(),
2499 Some(SqlType::Uuid) => row
2500 .try_get::<Option<Uuid>, _>(name)?
2501 .map_or(String::new(), |v| v.to_string()),
2502 _ => row
2503 .try_get::<Option<i64>, _>(name)?
2504 .map_or(String::new(), |v| v.to_string()),
2505 },
2506 SqlType::Bytes => row
2507 .try_get::<Option<Vec<u8>>, _>(name)?
2508 .map_or(String::new(), |b| hex_encode(&b)),
2509 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2510 });
2511 }
2512 Ok(match col.ty {
2513 SqlType::SmallInt | SqlType::Integer => row.try_get::<i32, _>(name)?.to_string(),
2514 SqlType::BigInt => row.try_get::<i64, _>(name)?.to_string(),
2515 SqlType::Real => row.try_get::<f32, _>(name)?.to_string(),
2516 SqlType::Double => row.try_get::<f64, _>(name)?.to_string(),
2517 SqlType::Boolean => if row.try_get::<bool, _>(name)? {
2518 "true"
2519 } else {
2520 "false"
2521 }
2522 .to_string(),
2523 SqlType::Text => row.try_get::<String, _>(name)?,
2524 SqlType::Date => row.try_get::<NaiveDate, _>(name)?.to_string(),
2525 SqlType::Time => row.try_get::<NaiveTime, _>(name)?.to_string(),
2526 SqlType::Timestamptz => row.try_get::<DateTime<Utc>, _>(name)?.to_rfc3339(),
2527 SqlType::Uuid => row.try_get::<Uuid, _>(name)?.to_string(),
2528 SqlType::Json => row.try_get::<Value, _>(name)?.to_string(),
2529 SqlType::Array(_) => panic_array_unsupported(&col.name),
2530 SqlType::Inet
2531 | SqlType::Cidr
2532 | SqlType::MacAddr
2533 | SqlType::Xml
2534 | SqlType::Ltree
2535 | SqlType::Bit
2536 | SqlType::FullText => panic_pg_only_unsupported(&col.name),
2537 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2538 Some(SqlType::Text) => row.try_get::<String, _>(name)?,
2539 Some(SqlType::Uuid) => row.try_get::<Uuid, _>(name)?.to_string(),
2540 _ => row.try_get::<i64, _>(name)?.to_string(),
2541 },
2542 SqlType::Bytes => hex_encode(&row.try_get::<Vec<u8>, _>(name)?),
2543 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2544 })
2545}
2546
2547pub fn decode_pg_to_string(
2558 row: &sqlx::postgres::PgRow,
2559 col: &Column,
2560) -> Result<String, sqlx::Error> {
2561 use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
2562 use serde_json::Value;
2563 use uuid::Uuid;
2564
2565 let name = col.name.as_str();
2566 if col.nullable {
2567 return Ok(match col.ty {
2568 SqlType::SmallInt => row
2569 .try_get::<Option<i16>, _>(name)?
2570 .map_or(String::new(), |v| v.to_string()),
2571 SqlType::Integer => row
2572 .try_get::<Option<i32>, _>(name)?
2573 .map_or(String::new(), |v| v.to_string()),
2574 SqlType::BigInt => row
2575 .try_get::<Option<i64>, _>(name)?
2576 .map_or(String::new(), |v| v.to_string()),
2577 SqlType::Real => row
2578 .try_get::<Option<f32>, _>(name)?
2579 .map_or(String::new(), |v| v.to_string()),
2580 SqlType::Double => row
2581 .try_get::<Option<f64>, _>(name)?
2582 .map_or(String::new(), |v| v.to_string()),
2583 SqlType::Boolean => row
2584 .try_get::<Option<bool>, _>(name)?
2585 .map_or(String::new(), |v| {
2586 if v { "true" } else { "false" }.to_string()
2587 }),
2588 SqlType::Text => row.try_get::<Option<String>, _>(name)?.unwrap_or_default(),
2589 SqlType::Date => row
2590 .try_get::<Option<NaiveDate>, _>(name)?
2591 .map_or(String::new(), |v| v.to_string()),
2592 SqlType::Time => row
2593 .try_get::<Option<NaiveTime>, _>(name)?
2594 .map_or(String::new(), |v| v.to_string()),
2595 SqlType::Timestamptz => row
2596 .try_get::<Option<DateTime<Utc>>, _>(name)?
2597 .map_or(String::new(), |v| v.to_rfc3339()),
2598 SqlType::Uuid => row
2599 .try_get::<Option<Uuid>, _>(name)?
2600 .map_or(String::new(), |v| v.to_string()),
2601 SqlType::Json => row
2602 .try_get::<Option<Value>, _>(name)?
2603 .map_or(String::new(), |v| v.to_string()),
2604 SqlType::Array(_)
2610 | SqlType::Inet
2611 | SqlType::Cidr
2612 | SqlType::MacAddr
2613 | SqlType::Xml
2614 | SqlType::Ltree
2615 | SqlType::Bit
2616 | SqlType::FullText => row
2617 .try_get::<Option<String>, _>(name)
2618 .ok()
2619 .flatten()
2620 .unwrap_or_default(),
2621 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2624 Some(SqlType::Text) => row.try_get::<Option<String>, _>(name)?.unwrap_or_default(),
2625 Some(SqlType::Uuid) => row
2626 .try_get::<Option<Uuid>, _>(name)?
2627 .map_or(String::new(), |v| v.to_string()),
2628 _ => row
2629 .try_get::<Option<i64>, _>(name)?
2630 .map_or(String::new(), |v| v.to_string()),
2631 },
2632 SqlType::Bytes => row
2633 .try_get::<Option<Vec<u8>>, _>(name)?
2634 .map_or(String::new(), |b| hex_encode(&b)),
2635 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2636 });
2637 }
2638 Ok(match col.ty {
2639 SqlType::SmallInt => row.try_get::<i16, _>(name)?.to_string(),
2640 SqlType::Integer => row.try_get::<i32, _>(name)?.to_string(),
2641 SqlType::BigInt => row.try_get::<i64, _>(name)?.to_string(),
2642 SqlType::Real => row.try_get::<f32, _>(name)?.to_string(),
2643 SqlType::Double => row.try_get::<f64, _>(name)?.to_string(),
2644 SqlType::Boolean => if row.try_get::<bool, _>(name)? {
2645 "true"
2646 } else {
2647 "false"
2648 }
2649 .to_string(),
2650 SqlType::Text => row.try_get::<String, _>(name)?,
2651 SqlType::Date => row.try_get::<NaiveDate, _>(name)?.to_string(),
2652 SqlType::Time => row.try_get::<NaiveTime, _>(name)?.to_string(),
2653 SqlType::Timestamptz => row.try_get::<DateTime<Utc>, _>(name)?.to_rfc3339(),
2654 SqlType::Uuid => row.try_get::<Uuid, _>(name)?.to_string(),
2655 SqlType::Json => row.try_get::<Value, _>(name)?.to_string(),
2656 SqlType::Array(_)
2658 | SqlType::Inet
2659 | SqlType::Cidr
2660 | SqlType::MacAddr
2661 | SqlType::Xml
2662 | SqlType::Ltree
2663 | SqlType::Bit
2664 | SqlType::FullText => row.try_get::<String, _>(name).unwrap_or_default(),
2665 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2666 Some(SqlType::Text) => row.try_get::<String, _>(name)?,
2667 Some(SqlType::Uuid) => row.try_get::<Uuid, _>(name)?.to_string(),
2668 _ => row.try_get::<i64, _>(name)?.to_string(),
2669 },
2670 SqlType::Bytes => hex_encode(&row.try_get::<Vec<u8>, _>(name)?),
2671 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2672 })
2673}
2674
2675pub fn decode_to_json_aliased(
2688 row: &sqlx::sqlite::SqliteRow,
2689 col: &Column,
2690 alias: &str,
2691) -> Result<serde_json::Value, sqlx::Error> {
2692 let mut aliased = col.clone();
2693 aliased.name = alias.to_string();
2694 decode_to_json(row, &aliased)
2695}
2696
2697pub fn decode_pg_to_json_aliased(
2699 row: &sqlx::postgres::PgRow,
2700 col: &Column,
2701 alias: &str,
2702) -> Result<serde_json::Value, sqlx::Error> {
2703 let mut aliased = col.clone();
2704 aliased.name = alias.to_string();
2705 decode_pg_to_json(row, &aliased)
2706}
2707
2708fn fk_target_pk_sql_type(col: &Column) -> Option<SqlType> {
2732 if !matches!(col.ty, SqlType::ForeignKey) {
2733 return None;
2734 }
2735 let target_table = col.fk_target.as_deref()?;
2736 crate::migrate::pk_meta_for_table(target_table).map(|(_, ty)| ty)
2737}
2738
2739pub fn decode_to_json(
2740 row: &sqlx::sqlite::SqliteRow,
2741 col: &Column,
2742) -> Result<serde_json::Value, sqlx::Error> {
2743 use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
2744 use serde_json::Value;
2745 use uuid::Uuid;
2746
2747 let name = col.name.as_str();
2748 if col.nullable {
2749 return Ok(match col.ty {
2750 SqlType::SmallInt | SqlType::Integer => row
2751 .try_get::<Option<i32>, _>(name)?
2752 .map_or(Value::Null, Value::from),
2753 SqlType::BigInt => row
2754 .try_get::<Option<i64>, _>(name)?
2755 .map_or(Value::Null, Value::from),
2756 SqlType::Real => row
2757 .try_get::<Option<f32>, _>(name)?
2758 .map_or(Value::Null, |v| Value::from(v as f64)),
2759 SqlType::Double => row
2760 .try_get::<Option<f64>, _>(name)?
2761 .map_or(Value::Null, Value::from),
2762 SqlType::Boolean => row
2763 .try_get::<Option<bool>, _>(name)?
2764 .map_or(Value::Null, Value::from),
2765 SqlType::Text => row
2766 .try_get::<Option<String>, _>(name)?
2767 .map_or(Value::Null, Value::from),
2768 SqlType::Date => row
2769 .try_get::<Option<NaiveDate>, _>(name)?
2770 .map_or(Value::Null, |v| Value::from(v.to_string())),
2771 SqlType::Time => row
2772 .try_get::<Option<NaiveTime>, _>(name)?
2773 .map_or(Value::Null, |v| Value::from(v.to_string())),
2774 SqlType::Timestamptz => row
2775 .try_get::<Option<DateTime<Utc>>, _>(name)?
2776 .map_or(Value::Null, |v| Value::from(v.to_rfc3339())),
2777 SqlType::Uuid => row
2778 .try_get::<Option<Uuid>, _>(name)?
2779 .map_or(Value::Null, |v| Value::from(v.to_string())),
2780 SqlType::Json => row
2781 .try_get::<Option<Value>, _>(name)?
2782 .unwrap_or(Value::Null),
2783 SqlType::Array(_) => panic_array_unsupported(&col.name),
2784 SqlType::Inet
2785 | SqlType::Cidr
2786 | SqlType::MacAddr
2787 | SqlType::Xml
2788 | SqlType::Ltree
2789 | SqlType::Bit
2790 | SqlType::FullText => panic_pg_only_unsupported(&col.name),
2791 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2796 Some(SqlType::Text) => row
2797 .try_get::<Option<String>, _>(name)?
2798 .map_or(Value::Null, Value::from),
2799 Some(SqlType::Uuid) => row
2800 .try_get::<Option<Uuid>, _>(name)?
2801 .map_or(Value::Null, |v| Value::from(v.to_string())),
2802 _ => row
2803 .try_get::<Option<i64>, _>(name)?
2804 .map_or(Value::Null, Value::from),
2805 },
2806 SqlType::Bytes => row
2807 .try_get::<Option<Vec<u8>>, _>(name)?
2808 .map_or(Value::Null, |b| bytes_to_json(&b)),
2809 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2810 });
2811 }
2812 Ok(match col.ty {
2813 SqlType::SmallInt | SqlType::Integer => Value::from(row.try_get::<i32, _>(name)?),
2814 SqlType::BigInt => Value::from(row.try_get::<i64, _>(name)?),
2815 SqlType::Real => Value::from(row.try_get::<f32, _>(name)? as f64),
2816 SqlType::Double => Value::from(row.try_get::<f64, _>(name)?),
2817 SqlType::Boolean => Value::from(row.try_get::<bool, _>(name)?),
2818 SqlType::Text => Value::from(row.try_get::<String, _>(name)?),
2819 SqlType::Date => Value::from(row.try_get::<NaiveDate, _>(name)?.to_string()),
2820 SqlType::Time => Value::from(row.try_get::<NaiveTime, _>(name)?.to_string()),
2821 SqlType::Timestamptz => Value::from(row.try_get::<DateTime<Utc>, _>(name)?.to_rfc3339()),
2822 SqlType::Uuid => Value::from(row.try_get::<Uuid, _>(name)?.to_string()),
2823 SqlType::Json => row.try_get::<Value, _>(name)?,
2824 SqlType::Array(_) => panic_array_unsupported(&col.name),
2825 SqlType::Inet
2826 | SqlType::Cidr
2827 | SqlType::MacAddr
2828 | SqlType::Xml
2829 | SqlType::Ltree
2830 | SqlType::Bit
2831 | SqlType::FullText => panic_pg_only_unsupported(&col.name),
2832 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2835 Some(SqlType::Text) => Value::from(row.try_get::<String, _>(name)?),
2836 Some(SqlType::Uuid) => Value::from(row.try_get::<Uuid, _>(name)?.to_string()),
2837 _ => Value::from(row.try_get::<i64, _>(name)?),
2838 },
2839 SqlType::Bytes => bytes_to_json(&row.try_get::<Vec<u8>, _>(name)?),
2840 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2841 })
2842}
2843
2844pub fn decode_pg_to_json(
2848 row: &sqlx::postgres::PgRow,
2849 col: &Column,
2850) -> Result<serde_json::Value, sqlx::Error> {
2851 use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
2852 use serde_json::Value;
2853 use uuid::Uuid;
2854
2855 let name = col.name.as_str();
2856 if col.nullable {
2857 return Ok(match col.ty {
2858 SqlType::SmallInt => row
2859 .try_get::<Option<i16>, _>(name)?
2860 .map_or(Value::Null, Value::from),
2861 SqlType::Integer => row
2862 .try_get::<Option<i32>, _>(name)?
2863 .map_or(Value::Null, Value::from),
2864 SqlType::BigInt => row
2865 .try_get::<Option<i64>, _>(name)?
2866 .map_or(Value::Null, Value::from),
2867 SqlType::Real => row
2868 .try_get::<Option<f32>, _>(name)?
2869 .map_or(Value::Null, |v| Value::from(v as f64)),
2870 SqlType::Double => row
2871 .try_get::<Option<f64>, _>(name)?
2872 .map_or(Value::Null, Value::from),
2873 SqlType::Boolean => row
2874 .try_get::<Option<bool>, _>(name)?
2875 .map_or(Value::Null, Value::from),
2876 SqlType::Text => row
2877 .try_get::<Option<String>, _>(name)?
2878 .map_or(Value::Null, Value::from),
2879 SqlType::Date => row
2880 .try_get::<Option<NaiveDate>, _>(name)?
2881 .map_or(Value::Null, |v| Value::from(v.to_string())),
2882 SqlType::Time => row
2883 .try_get::<Option<NaiveTime>, _>(name)?
2884 .map_or(Value::Null, |v| Value::from(v.to_string())),
2885 SqlType::Timestamptz => row
2886 .try_get::<Option<DateTime<Utc>>, _>(name)?
2887 .map_or(Value::Null, |v| Value::from(v.to_rfc3339())),
2888 SqlType::Uuid => row
2889 .try_get::<Option<Uuid>, _>(name)?
2890 .map_or(Value::Null, |v| Value::from(v.to_string())),
2891 SqlType::Json => row
2892 .try_get::<Option<Value>, _>(name)?
2893 .unwrap_or(Value::Null),
2894 SqlType::Array(_)
2895 | SqlType::Inet
2896 | SqlType::Cidr
2897 | SqlType::MacAddr
2898 | SqlType::Xml
2899 | SqlType::Ltree
2900 | SqlType::Bit
2901 | SqlType::FullText => row
2902 .try_get::<Option<String>, _>(name)
2903 .ok()
2904 .flatten()
2905 .map_or(Value::Null, Value::from),
2906 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2909 Some(SqlType::Text) => row
2910 .try_get::<Option<String>, _>(name)?
2911 .map_or(Value::Null, Value::from),
2912 Some(SqlType::Uuid) => row
2913 .try_get::<Option<Uuid>, _>(name)?
2914 .map_or(Value::Null, |v| Value::from(v.to_string())),
2915 _ => row
2916 .try_get::<Option<i64>, _>(name)?
2917 .map_or(Value::Null, Value::from),
2918 },
2919 SqlType::Bytes => row
2920 .try_get::<Option<Vec<u8>>, _>(name)?
2921 .map_or(Value::Null, |b| bytes_to_json(&b)),
2922 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2923 });
2924 }
2925 Ok(match col.ty {
2926 SqlType::SmallInt => Value::from(row.try_get::<i16, _>(name)?),
2927 SqlType::Integer => Value::from(row.try_get::<i32, _>(name)?),
2928 SqlType::BigInt => Value::from(row.try_get::<i64, _>(name)?),
2929 SqlType::Real => Value::from(row.try_get::<f32, _>(name)? as f64),
2930 SqlType::Double => Value::from(row.try_get::<f64, _>(name)?),
2931 SqlType::Boolean => Value::from(row.try_get::<bool, _>(name)?),
2932 SqlType::Text => Value::from(row.try_get::<String, _>(name)?),
2933 SqlType::Date => Value::from(row.try_get::<NaiveDate, _>(name)?.to_string()),
2934 SqlType::Time => Value::from(row.try_get::<NaiveTime, _>(name)?.to_string()),
2935 SqlType::Timestamptz => Value::from(row.try_get::<DateTime<Utc>, _>(name)?.to_rfc3339()),
2936 SqlType::Uuid => Value::from(row.try_get::<Uuid, _>(name)?.to_string()),
2937 SqlType::Json => row.try_get::<Value, _>(name)?,
2938 SqlType::Array(_)
2939 | SqlType::Inet
2940 | SqlType::Cidr
2941 | SqlType::MacAddr
2942 | SqlType::Xml
2943 | SqlType::Ltree
2944 | SqlType::Bit
2945 | SqlType::FullText => row
2946 .try_get::<String, _>(name)
2947 .map(Value::from)
2948 .unwrap_or(Value::Null),
2949 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2952 Some(SqlType::Text) => Value::from(row.try_get::<String, _>(name)?),
2953 Some(SqlType::Uuid) => Value::from(row.try_get::<Uuid, _>(name)?.to_string()),
2954 _ => Value::from(row.try_get::<i64, _>(name)?),
2955 },
2956 SqlType::Bytes => bytes_to_json(&row.try_get::<Vec<u8>, _>(name)?),
2957 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2958 })
2959}
2960
2961fn normalize_str(col: &Column, s: &str) -> String {
2967 let trimmed = if col.trim { s.trim() } else { s };
2968 if col.lowercase {
2969 trimmed.to_lowercase()
2970 } else {
2971 trimmed.to_string()
2972 }
2973}
2974
2975fn normalize_json_for_col(col: &Column, v: &serde_json::Value) -> Option<serde_json::Value> {
2981 if !(col.trim || col.lowercase) {
2982 return None;
2983 }
2984 let s = v.as_str()?;
2985 Some(serde_json::Value::String(normalize_str(col, s)))
2986}
2987
2988fn form_str_to_sea_value(col: &Column, raw: &str) -> Result<SeaValue, WriteError> {
2996 let normalized = normalize_str(col, raw);
3000 let raw = normalized.as_str();
3001 if raw.is_empty() {
3002 if col.ty == SqlType::Boolean {
3003 return Ok(SeaValue::Bool(Some(false)));
3005 }
3006 if col.nullable {
3007 return Ok(null_for(col.ty));
3008 }
3009 return Err(WriteError::RequiredFieldMissing {
3010 field: col.name.clone(),
3011 });
3012 }
3013 if matches!(col.ty, SqlType::Json | SqlType::Array(_)) {
3025 let parsed: serde_json::Value =
3026 serde_json::from_str(raw).map_err(|e| WriteError::Validator {
3027 field: col.name.clone(),
3028 message: format!("Not valid JSON: {e}"),
3029 })?;
3030 return json_to_sea_value(col.ty, &parsed, col.nullable, &col.name, None);
3031 }
3032 if matches!(col.ty, SqlType::ForeignKey) {
3033 return match fk_target_pk_sql_type(col) {
3034 Some(SqlType::Text) => Ok(SeaValue::String(Some(Box::new(raw.to_string())))),
3035 Some(SqlType::Uuid) => uuid::Uuid::parse_str(raw)
3036 .map(|v| SeaValue::Uuid(Some(Box::new(v))))
3037 .map_err(|_| WriteError::TypeMismatch {
3038 field: col.name.clone(),
3039 expected: SqlType::Uuid,
3040 got: raw.to_string(),
3041 }),
3042 _ => raw
3043 .parse::<i64>()
3044 .map(|v| SeaValue::BigInt(Some(v)))
3045 .map_err(|_| WriteError::TypeMismatch {
3046 field: col.name.clone(),
3047 expected: SqlType::BigInt,
3048 got: raw.to_string(),
3049 }),
3050 };
3051 }
3052 let json = serde_json::Value::String(raw.to_string());
3053 if let Some(sealed) = crate::orm::write::seal_masked_json(col, &json)? {
3056 return json_to_sea_value(col.ty, &sealed, col.nullable, &col.name, None);
3057 }
3058 json_to_sea_value(col.ty, &json, col.nullable, &col.name, None)
3059}
3060
3061fn hex_encode(bytes: &[u8]) -> String {
3065 let mut out = String::with_capacity(bytes.len() * 2);
3066 for b in bytes {
3067 out.push_str(&format!("{b:02x}"));
3068 }
3069 out
3070}
3071
3072fn bytes_to_json(bytes: &[u8]) -> serde_json::Value {
3075 serde_json::Value::Array(bytes.iter().map(|b| serde_json::Value::from(*b)).collect())
3076}
3077
3078fn panic_array_unsupported(column: &str) -> ! {
3079 panic!(
3080 "DynQuerySet: column `{column}` is a Postgres-only Array; the \
3081 field/backend system check should have failed boot."
3082 )
3083}
3084
3085fn panic_pg_only_unsupported(column: &str) -> ! {
3086 panic!(
3087 "DynQuerySet: column `{column}` is a Postgres-only network type \
3088 (Inet/Cidr/MacAddr); the field/backend system check should \
3089 have failed boot."
3090 )
3091}
3092
3093fn classify_or_sqlx(
3099 e: sqlx::Error,
3100 body: &serde_json::Map<String, serde_json::Value>,
3101) -> crate::orm::write::WriteError {
3102 if let Some(classified) = crate::orm::validation::classify_sql_error(&e, body) {
3103 return classified;
3104 }
3105 crate::orm::write::WriteError::Sqlx(e)
3106}
3107
3108fn validate_numeric_bounds(
3109 col: &Column,
3110 json: &serde_json::Value,
3111) -> Result<(), crate::orm::write::WriteError> {
3112 let Some(n) = json.as_f64() else {
3113 return Ok(());
3114 };
3115 if let Some(min) = col.min {
3116 if n < min as f64 {
3117 return Err(crate::orm::write::WriteError::Validator {
3118 field: col.name.clone(),
3119 message: format!("must be >= {min} (got {n})."),
3120 });
3121 }
3122 }
3123 if let Some(max) = col.max {
3124 if n > max as f64 {
3125 return Err(crate::orm::write::WriteError::Validator {
3126 field: col.name.clone(),
3127 message: format!("must be <= {max} (got {n})."),
3128 });
3129 }
3130 }
3131 Ok(())
3132}
3133
3134fn json_pk_to_sea(v: &serde_json::Value) -> Option<sea_query::Value> {
3140 match v {
3141 serde_json::Value::Number(n) => n.as_i64().map(|i| sea_query::Value::BigInt(Some(i))),
3142 serde_json::Value::String(s) => Some(sea_query::Value::String(Some(Box::new(s.clone())))),
3143 _ => None,
3144 }
3145}
3146
3147fn normalize_sr_token(name: &str) -> String {
3164 name.replace("__", ".")
3165}
3166
3167fn validate_sr_chain(root_meta: &crate::migrate::ModelMeta, chain: &str) -> Option<Vec<String>> {
3177 let hops: Vec<&str> = chain.split('.').filter(|s| !s.is_empty()).collect();
3178 if hops.is_empty() {
3179 return None;
3180 }
3181 let registered = crate::migrate::registered_models();
3182 let mut targets: Vec<String> = Vec::with_capacity(hops.len());
3183 let mut current_table: String = root_meta.table.clone();
3184 let mut current_meta: Option<crate::migrate::ModelMeta> = None;
3185 for hop in &hops {
3186 let meta_ref: &crate::migrate::ModelMeta =
3187 if current_table == root_meta.table && current_meta.is_none() {
3188 root_meta
3189 } else {
3190 current_meta = registered
3191 .iter()
3192 .find(|m| m.table == current_table)
3193 .cloned();
3194 current_meta.as_ref()?
3195 };
3196 let col = meta_ref.fields.iter().find(|c| &c.name == hop)?;
3197 let target = col.fk_target.clone()?;
3198 targets.push(target.clone());
3199 current_table = target;
3200 }
3201 Some(targets)
3202}
3203
3204async fn hydrate_select_related_into(
3221 meta: &crate::migrate::ModelMeta,
3222 sr_fields: &[String],
3223 rows: &mut [serde_json::Map<String, serde_json::Value>],
3224) -> Result<(), sqlx::Error> {
3225 let pool = resolve_pool_dyn(meta, crate::db::RouteOp::Read);
3226 for chain in sr_fields {
3227 let hops: Vec<&str> = chain.split('.').filter(|s| !s.is_empty()).collect();
3228 if hops.is_empty() {
3229 continue;
3230 }
3231 let Some(targets) = validate_sr_chain(meta, chain) else {
3232 continue;
3237 };
3238
3239 let registered = crate::migrate::registered_models();
3248 let hop_target_pk: Vec<(String, SqlType)> = targets
3249 .iter()
3250 .filter_map(|t| {
3251 registered
3252 .iter()
3253 .find(|m| &m.table == t)
3254 .and_then(|m| m.pk_column().map(|c| (c.name.clone(), c.ty)))
3255 })
3256 .collect();
3257 if hop_target_pk.len() != hops.len() {
3258 continue;
3262 }
3263 let hop_target_soft_delete: Vec<bool> = targets
3264 .iter()
3265 .map(|t| {
3266 registered
3267 .iter()
3268 .find(|m| &m.table == t)
3269 .is_some_and(|m| m.soft_delete)
3270 })
3271 .collect();
3272
3273 let first_field = hops[0];
3277 let mut ids: Vec<serde_json::Value> = Vec::with_capacity(rows.len());
3278 for row in rows.iter() {
3279 let Some(v) = row.get(first_field) else {
3280 continue;
3281 };
3282 if v.is_null() {
3283 continue;
3284 }
3285 ids.push(v.clone());
3286 }
3287 if ids.is_empty() {
3288 continue;
3289 }
3290 dedup_by_pk_key(&mut ids);
3291 let mut levels: Vec<Vec<serde_json::Value>> = Vec::with_capacity(hops.len());
3292 levels.push(
3293 crate::orm::queryset::hydration::fetch_related_as_json_by_pk(
3294 &targets[0],
3295 &hop_target_pk[0].0,
3296 hop_target_pk[0].1,
3297 hop_target_soft_delete[0],
3298 &ids,
3299 &pool,
3300 )
3301 .await?,
3302 );
3303
3304 for hop_idx in 1..hops.len() {
3305 let hop_field = hops[hop_idx];
3306 let hop_target = &targets[hop_idx];
3307 let prev_lvl = &levels[hop_idx - 1];
3308 let mut next_ids: Vec<serde_json::Value> = prev_lvl
3309 .iter()
3310 .filter_map(|r| {
3311 let v = r.as_object()?.get(hop_field)?;
3312 if v.is_null() { None } else { Some(v.clone()) }
3313 })
3314 .collect();
3315 if next_ids.is_empty() {
3316 break;
3321 }
3322 dedup_by_pk_key(&mut next_ids);
3323 levels.push(
3324 crate::orm::queryset::hydration::fetch_related_as_json_by_pk(
3325 hop_target,
3326 &hop_target_pk[hop_idx].0,
3327 hop_target_pk[hop_idx].1,
3328 hop_target_soft_delete[hop_idx],
3329 &next_ids,
3330 &pool,
3331 )
3332 .await?,
3333 );
3334 }
3335
3336 if levels.len() > 1 {
3341 for i in (0..levels.len() - 1).rev() {
3342 let next_pk_col = &hop_target_pk[i + 1].0;
3343 let next_by_pk: HashMap<String, serde_json::Value> = levels[i + 1]
3344 .iter()
3345 .filter_map(|obj| {
3346 let map = obj.as_object()?;
3347 let pk_val = map.get(next_pk_col.as_str())?;
3348 Some((pk_json_key(pk_val), obj.clone()))
3349 })
3350 .collect();
3351 let hop_field = hops[i + 1];
3352 for row in levels[i].iter_mut() {
3353 let Some(map) = row.as_object_mut() else {
3354 continue;
3355 };
3356 let Some(fk_val) = map.get(hop_field) else {
3357 continue;
3358 };
3359 if fk_val.is_null() {
3360 continue;
3361 }
3362 let key = pk_json_key(fk_val);
3363 if let Some(next_json) = next_by_pk.get(&key) {
3364 map.insert(hop_field.to_string(), next_json.clone());
3365 }
3366 }
3367 }
3368 }
3369
3370 let first_pk_col = &hop_target_pk[0].0;
3377 let first_by_pk: HashMap<String, serde_json::Value> = levels
3378 .into_iter()
3379 .next()
3380 .unwrap_or_default()
3381 .into_iter()
3382 .filter_map(|obj| {
3383 let map = obj.as_object()?;
3384 let pk_val = map.get(first_pk_col.as_str())?;
3385 Some((pk_json_key(pk_val), obj.clone()))
3386 })
3387 .collect();
3388 for row in rows.iter_mut() {
3389 let Some(fk_val) = row.get(first_field) else {
3390 continue;
3391 };
3392 if fk_val.is_null() {
3393 continue;
3394 }
3395 let key = pk_json_key(fk_val);
3396 if let Some(resolved) = first_by_pk.get(&key) {
3397 row.insert(first_field.to_string(), resolved.clone());
3398 }
3399 }
3400 }
3401 Ok(())
3402}
3403
3404fn dedup_by_pk_key(ids: &mut Vec<serde_json::Value>) {
3409 let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
3410 ids.retain(|v| seen.insert(pk_json_key(v)));
3411}
3412
3413async fn hydrate_m2m_batched(
3427 meta: &crate::migrate::ModelMeta,
3428 pk_name: &str,
3429 rows: &mut [serde_json::Map<String, serde_json::Value>],
3430) -> Result<(), sqlx::Error> {
3431 if meta.m2m_relations.is_empty() || rows.is_empty() {
3432 return Ok(());
3433 }
3434
3435 for row in rows.iter_mut() {
3440 for rel in &meta.m2m_relations {
3441 row.insert(rel.field_name.clone(), serde_json::Value::Array(Vec::new()));
3442 }
3443 }
3444
3445 let mut parent_sea_vals: Vec<sea_query::Value> = Vec::with_capacity(rows.len());
3449 let mut seen_keys: std::collections::HashSet<String> = std::collections::HashSet::new();
3450 for row in rows.iter() {
3451 let Some(pk_json) = row.get(pk_name) else {
3452 continue;
3453 };
3454 let Some(sea_val) = json_pk_to_sea(pk_json) else {
3455 continue;
3456 };
3457 let key = pk_json_key(pk_json);
3458 if seen_keys.insert(key) {
3459 parent_sea_vals.push(sea_val);
3460 }
3461 }
3462 if parent_sea_vals.is_empty() {
3463 return Ok(());
3464 }
3465
3466 for rel in &meta.m2m_relations {
3467 let junction_table = format!("{}_{}", meta.table, rel.field_name);
3468 let mut sel = Query::select();
3469 sel.from(crate::db::router::schema_qualified_table(&junction_table));
3470 sel.column(Alias::new("parent_id"));
3471 sel.column(Alias::new("child_id"));
3472 sel.and_where(Expr::col(Alias::new("parent_id")).is_in(parent_sea_vals.clone()));
3473
3474 let mut children_by_parent: HashMap<String, Vec<serde_json::Value>> = HashMap::new();
3475 match resolve_pool_dyn(meta, crate::db::RouteOp::Read) {
3476 DbPool::Sqlite(pool) => {
3477 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
3478 let db_rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
3479 for r in &db_rows {
3480 let parent = read_junction_id_sqlite(r, "parent_id")?;
3481 let child = read_junction_id_sqlite(r, "child_id")?;
3482 children_by_parent
3483 .entry(pk_json_key(&parent))
3484 .or_default()
3485 .push(child);
3486 }
3487 }
3488 DbPool::Postgres(pool) => {
3489 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
3490 let db_rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
3491 for r in &db_rows {
3492 let parent = read_junction_id_pg(r, "parent_id")?;
3493 let child = read_junction_id_pg(r, "child_id")?;
3494 children_by_parent
3495 .entry(pk_json_key(&parent))
3496 .or_default()
3497 .push(child);
3498 }
3499 }
3500 }
3501
3502 for row in rows.iter_mut() {
3503 let Some(pk_json) = row.get(pk_name) else {
3504 continue;
3505 };
3506 let key = pk_json_key(pk_json);
3507 if let Some(children) = children_by_parent.remove(&key) {
3508 row.insert(rel.field_name.clone(), serde_json::Value::Array(children));
3509 }
3510 }
3511 }
3512 Ok(())
3513}
3514
3515fn pk_json_key(v: &serde_json::Value) -> String {
3521 match v {
3522 serde_json::Value::Number(n) => format!("n:{n}"),
3523 serde_json::Value::String(s) => format!("s:{s}"),
3524 other => format!("o:{other}"),
3525 }
3526}
3527
3528fn read_junction_id_sqlite(
3533 row: &sqlx::sqlite::SqliteRow,
3534 col: &str,
3535) -> Result<serde_json::Value, sqlx::Error> {
3536 if let Ok(i) = row.try_get::<i64, _>(col) {
3537 return Ok(serde_json::Value::Number(i.into()));
3538 }
3539 let s = row.try_get::<String, _>(col)?;
3540 Ok(serde_json::Value::String(s))
3541}
3542
3543fn read_junction_id_pg(
3544 row: &sqlx::postgres::PgRow,
3545 col: &str,
3546) -> Result<serde_json::Value, sqlx::Error> {
3547 if let Ok(i) = row.try_get::<i64, _>(col) {
3548 return Ok(serde_json::Value::Number(i.into()));
3549 }
3550 let s = row.try_get::<String, _>(col)?;
3551 Ok(serde_json::Value::String(s))
3552}
3553
3554pub(crate) async fn audit_snapshot(
3563 meta: &crate::migrate::ModelMeta,
3564 clauses: &[Condition],
3565) -> Vec<serde_json::Map<String, serde_json::Value>> {
3566 let mut qs = DynQuerySet::for_meta(meta);
3567 for c in clauses {
3568 qs = qs.filter_condition(c.clone());
3569 }
3570 qs.fetch_as_json().await.unwrap_or_default()
3571}
3572
3573#[allow(clippy::type_complexity)]
3575pub(crate) fn audit_pairs(
3576 meta: &crate::migrate::ModelMeta,
3577 before: Vec<serde_json::Map<String, serde_json::Value>>,
3578 after: Vec<serde_json::Map<String, serde_json::Value>>,
3579) -> Vec<(
3580 String,
3581 Option<serde_json::Map<String, serde_json::Value>>,
3582 Option<serde_json::Map<String, serde_json::Value>>,
3583)> {
3584 use crate::orm::audit;
3585 let mut by_pk: std::collections::HashMap<String, serde_json::Map<String, serde_json::Value>> =
3586 after
3587 .into_iter()
3588 .map(|r| (audit::pk_of(meta, &r), r))
3589 .collect();
3590 before
3591 .into_iter()
3592 .map(|b| {
3593 let pk = audit::pk_of(meta, &b);
3594 let a = by_pk.remove(&pk);
3595 (pk, Some(b), a)
3596 })
3597 .collect()
3598}
3599
3600async fn collect_parent_pks(
3601 meta: &crate::migrate::ModelMeta,
3602 pk_col: &crate::migrate::Column,
3603 where_clauses: &[Condition],
3604) -> Result<Vec<serde_json::Value>, crate::orm::write::WriteError> {
3605 let mut sel = Query::select();
3606 sel.from(crate::db::router::schema_qualified_table(&meta.table));
3607 sel.column(Alias::new(&pk_col.name));
3608 for cond in where_clauses {
3609 sel.cond_where(cond.clone());
3610 }
3611 match resolve_pool_dyn(meta, crate::db::RouteOp::Read) {
3612 DbPool::Sqlite(pool) => {
3613 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
3614 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
3615 rows.iter()
3616 .map(|row| decode_to_json(row, pk_col))
3617 .collect::<Result<Vec<_>, _>>()
3618 .map_err(crate::orm::write::WriteError::Sqlx)
3619 }
3620 DbPool::Postgres(pool) => {
3621 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
3622 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
3623 rows.iter()
3624 .map(|row| decode_pg_to_json(row, pk_col))
3625 .collect::<Result<Vec<_>, _>>()
3626 .map_err(crate::orm::write::WriteError::Sqlx)
3627 }
3628 }
3629}
3630
3631async fn collect_parent_pks_in_tx(
3635 meta: &crate::migrate::ModelMeta,
3636 pk_col: &crate::migrate::Column,
3637 where_clauses: &[Condition],
3638 tx: &mut crate::db::Transaction,
3639) -> Result<Vec<serde_json::Value>, crate::orm::write::WriteError> {
3640 let mut sel = Query::select();
3641 sel.from(crate::db::router::schema_qualified_table(&meta.table));
3642 sel.column(Alias::new(&pk_col.name));
3643 for cond in where_clauses {
3644 sel.cond_where(cond.clone());
3645 }
3646 match tx.backend_name() {
3647 "sqlite" => {
3648 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
3649 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
3650 let rows = sqlx::query_with(&sql, values)
3651 .fetch_all(&mut **inner)
3652 .await?;
3653 rows.iter()
3654 .map(|row| decode_to_json(row, pk_col))
3655 .collect::<Result<Vec<_>, _>>()
3656 .map_err(crate::orm::write::WriteError::Sqlx)
3657 }
3658 _ => {
3659 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
3660 let inner = tx.as_pg_mut().expect("postgres backend_name");
3661 let rows = sqlx::query_with(&sql, values)
3662 .fetch_all(&mut **inner)
3663 .await?;
3664 rows.iter()
3665 .map(|row| decode_pg_to_json(row, pk_col))
3666 .collect::<Result<Vec<_>, _>>()
3667 .map_err(crate::orm::write::WriteError::Sqlx)
3668 }
3669 }
3670}
3671
3672fn is_unauthorized_privileged(col: &crate::migrate::Column, allow_privileged: &[String]) -> bool {
3691 col.privileged && !allow_privileged.iter().any(|a| a == &col.name)
3692}
3693
3694fn normalise_insert_body(
3695 meta: &crate::migrate::ModelMeta,
3696 body: &serde_json::Map<String, serde_json::Value>,
3697 allow_privileged: &[String],
3698) -> Option<serde_json::Map<String, serde_json::Value>> {
3699 let needs_owned = meta.fields.iter().any(|c| {
3700 c.noform || c.slug_from.is_some() || is_unauthorized_privileged(c, allow_privileged)
3701 });
3702 if !needs_owned {
3703 return None;
3704 }
3705 let mut owned = body.clone();
3706 for col in &meta.fields {
3707 if col.noform || is_unauthorized_privileged(col, allow_privileged) {
3708 owned.remove(&col.name);
3709 }
3710 }
3711 crate::orm::write::apply_slug_from(&meta.fields, &mut owned, false);
3712 Some(owned)
3713}
3714
3715fn normalise_update_body(
3720 meta: &crate::migrate::ModelMeta,
3721 body: &serde_json::Map<String, serde_json::Value>,
3722 allow_privileged: &[String],
3723) -> Option<serde_json::Map<String, serde_json::Value>> {
3724 let needs_owned = meta.fields.iter().any(|c| {
3725 c.noform || c.slug_from.is_some() || is_unauthorized_privileged(c, allow_privileged)
3726 });
3727 if !needs_owned {
3728 return None;
3729 }
3730 let mut owned = body.clone();
3731 for col in &meta.fields {
3732 if col.noform || is_unauthorized_privileged(col, allow_privileged) {
3733 owned.remove(&col.name);
3734 }
3735 }
3736 crate::orm::write::apply_slug_from(&meta.fields, &mut owned, true);
3737 Some(owned)
3738}
3739
3740struct InsertPlan {
3742 q: sea_query::InsertStatement,
3743 pk_name: String,
3744 pk_ty: SqlType,
3745}
3746
3747fn build_insert_plan(
3756 meta: &crate::migrate::ModelMeta,
3757 body: &serde_json::Map<String, serde_json::Value>,
3758) -> Result<InsertPlan, crate::orm::write::WriteError> {
3759 use crate::orm::write::{WriteError, is_default_pk};
3760
3761 let mut cols: Vec<&str> = Vec::new();
3762 let mut values: Vec<SeaValue> = Vec::new();
3763 for col in &meta.fields {
3764 if col.primary_key {
3765 let supplied = body.get(&col.name);
3766 let is_sentinel = match supplied {
3767 None | Some(serde_json::Value::Null) => true,
3768 Some(v) => is_default_pk(col.ty, v),
3769 };
3770 if matches!(
3771 col.ty,
3772 SqlType::Integer | SqlType::BigInt | SqlType::SmallInt
3773 ) && is_sentinel
3774 {
3775 continue;
3776 }
3777 }
3778 if col.auto_user_add || col.auto_user {
3782 cols.push(&col.name);
3783 values.push(crate::orm::write::user_for_column(col.ty));
3784 continue;
3785 }
3786 let Some(json) = body.get(&col.name) else {
3787 if col.auto_now_add || col.auto_now {
3788 let now_value = crate::orm::write::now_for_column(col.ty);
3789 cols.push(&col.name);
3790 values.push(now_value);
3791 continue;
3792 }
3793 continue;
3794 };
3795 if json.is_null() {
3796 continue;
3797 }
3798 validate_numeric_bounds(col, json)?;
3799 if let (Some(fmt), Some(s)) = (col.text_format.as_deref(), json.as_str()) {
3800 if let Err(e) = crate::orm::validators::validate_text_format(fmt, s) {
3801 return Err(WriteError::Validator {
3802 field: col.name.clone(),
3803 message: e.to_string(),
3804 });
3805 }
3806 }
3807 let normalized_json = normalize_json_for_col(col, json);
3810 let json = normalized_json.as_ref().unwrap_or(json);
3811 let cleaned_json = crate::orm::cleaners::apply(&meta.table, &col.name, json)?;
3813 let json = cleaned_json.as_ref().unwrap_or(json);
3814 let sealed = crate::orm::write::seal_masked_json(col, json)?;
3816 let sea_value = crate::orm::write::json_to_sea_value(
3817 col.ty,
3818 sealed.as_ref().unwrap_or(json),
3819 col.nullable,
3820 &col.name,
3821 fk_target_pk_sql_type(col),
3822 )?;
3823 cols.push(&col.name);
3824 values.push(sea_value);
3825 }
3826
3827 let pk_col = meta.fields.iter().find(|c| c.primary_key).ok_or_else(|| {
3828 WriteError::Sqlx(sqlx::Error::Protocol(
3829 "insert_json: model has no PK".to_string(),
3830 ))
3831 })?;
3832 let pk_name = pk_col.name.clone();
3833 let pk_ty = pk_col.ty;
3834
3835 let mut q = Query::insert();
3836 q.into_table(crate::db::router::schema_qualified_table(&meta.table));
3837 q.columns(cols.iter().map(|c| Alias::new(*c)).collect::<Vec<_>>());
3838 let exprs: Vec<sea_query::SimpleExpr> = values.into_iter().map(Into::into).collect();
3839 q.values_panic(exprs);
3840
3841 Ok(InsertPlan { q, pk_name, pk_ty })
3842}
3843
3844async fn write_m2m_junctions_in_tx(
3848 meta: &crate::migrate::ModelMeta,
3849 parent_pk_json: Option<&serde_json::Value>,
3850 body: &serde_json::Map<String, serde_json::Value>,
3851 tx: &mut crate::db::Transaction,
3852) -> Result<(), crate::orm::write::WriteError> {
3853 if meta.m2m_relations.is_empty() {
3854 return Ok(());
3855 }
3856 let Some(parent_pk_value) = parent_pk_json.and_then(json_pk_to_sea) else {
3857 return Ok(());
3858 };
3859 for rel in &meta.m2m_relations {
3860 let Some(value) = body.get(&rel.field_name) else {
3861 continue;
3862 };
3863 let Some(items) = value.as_array() else {
3864 continue;
3865 };
3866 let mut child_ids: Vec<sea_query::Value> = Vec::with_capacity(items.len());
3867 for item in items {
3868 if item.is_null() {
3869 continue;
3870 }
3871 if let Some(v) = json_pk_to_sea(item) {
3872 child_ids.push(v);
3873 }
3874 }
3875 let junction_table = format!("{}_{}", meta.table, rel.field_name);
3876 crate::orm::m2m::set_junction_dynamic_in_tx(
3877 &junction_table,
3878 parent_pk_value.clone(),
3879 child_ids,
3880 tx,
3881 )
3882 .await
3883 .map_err(crate::orm::write::WriteError::Sqlx)?;
3884 }
3885 Ok(())
3886}
3887
3888async fn hydrate_m2m_into_tx(
3893 meta: &crate::migrate::ModelMeta,
3894 parent_pk_json: Option<&serde_json::Value>,
3895 out: &mut serde_json::Map<String, serde_json::Value>,
3896 tx: &mut crate::db::Transaction,
3897) -> Result<(), sqlx::Error> {
3898 if meta.m2m_relations.is_empty() {
3899 return Ok(());
3900 }
3901 let Some(parent_pk_value) = parent_pk_json.and_then(json_pk_to_sea) else {
3902 return Ok(());
3903 };
3904 for rel in &meta.m2m_relations {
3905 let junction_table = format!("{}_{}", meta.table, rel.field_name);
3906 let mut sel = Query::select();
3907 sel.from(crate::db::router::schema_qualified_table(&junction_table));
3908 sel.column(Alias::new("child_id"));
3909 sel.and_where(Expr::col(Alias::new("parent_id")).eq(parent_pk_value.clone()));
3910 let children: Vec<serde_json::Value> = match tx.backend_name() {
3911 "sqlite" => {
3912 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
3913 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
3914 let rows = sqlx::query_with(&sql, values)
3915 .fetch_all(&mut **inner)
3916 .await?;
3917 rows.iter()
3918 .map(|r| {
3919 r.try_get::<i64, _>("child_id")
3920 .map(|i| serde_json::Value::Number(i.into()))
3921 .or_else(|_| {
3922 r.try_get::<String, _>("child_id")
3923 .map(serde_json::Value::String)
3924 })
3925 })
3926 .collect::<Result<Vec<_>, _>>()?
3927 }
3928 _ => {
3929 let inner = tx.as_pg_mut().expect("postgres backend_name");
3930 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
3931 let rows = sqlx::query_with(&sql, values)
3932 .fetch_all(&mut **inner)
3933 .await?;
3934 rows.iter()
3935 .map(|r| {
3936 r.try_get::<i64, _>("child_id")
3937 .map(|i| serde_json::Value::Number(i.into()))
3938 .or_else(|_| {
3939 r.try_get::<String, _>("child_id")
3940 .map(serde_json::Value::String)
3941 })
3942 })
3943 .collect::<Result<Vec<_>, _>>()?
3944 }
3945 };
3946 out.insert(rel.field_name.clone(), serde_json::Value::Array(children));
3947 }
3948 Ok(())
3949}
3950
3951fn coerce_csv_cell(col: &Column, raw: &str) -> serde_json::Value {
3968 use serde_json::Value;
3969 if raw.is_empty() && col.nullable {
3970 return Value::Null;
3971 }
3972 let ty = crate::migrate::fk_effective_type(col);
3979 match ty {
3980 SqlType::SmallInt | SqlType::Integer | SqlType::BigInt | SqlType::ForeignKey => raw
3981 .parse::<i64>()
3982 .map(Value::from)
3983 .unwrap_or_else(|_| Value::String(raw.to_string())),
3984 SqlType::Real | SqlType::Double => raw
3985 .parse::<f64>()
3986 .ok()
3987 .and_then(serde_json::Number::from_f64)
3988 .map(Value::Number)
3989 .unwrap_or_else(|| Value::String(raw.to_string())),
3990 SqlType::Boolean => match raw.trim().to_ascii_lowercase().as_str() {
3991 "true" | "1" | "t" | "yes" | "y" => Value::Bool(true),
3992 "false" | "0" | "f" | "no" | "n" => Value::Bool(false),
3993 _ => Value::String(raw.to_string()),
3994 },
3995 SqlType::Json => {
3996 serde_json::from_str(raw).unwrap_or_else(|_| Value::String(raw.to_string()))
3997 }
3998 _ => Value::String(raw.to_string()),
3999 }
4000}
4001
4002#[derive(Debug, Default)]
4009pub struct CsvImportReport {
4010 pub inserted: usize,
4011 pub errors: Vec<(usize, String)>,
4012}
4013
4014pub async fn import_table_rows(
4025 meta: &ModelMeta,
4026 headers: &[String],
4027 rows: &[Vec<String>],
4028) -> CsvImportReport {
4029 let col_for: HashMap<&str, &Column> =
4030 meta.fields.iter().map(|c| (c.name.as_str(), c)).collect();
4031
4032 let mut report = CsvImportReport::default();
4033 for (i, row) in rows.iter().enumerate() {
4034 let mut obj = serde_json::Map::new();
4035 for (header, cell) in headers.iter().zip(row.iter()) {
4036 if let Some(col) = col_for.get(header.as_str()) {
4037 obj.insert(header.clone(), coerce_csv_cell(col, cell));
4038 }
4039 }
4040 match DynQuerySet::for_meta(meta).insert_json(&obj).await {
4041 Ok(_) => report.inserted += 1,
4042 Err(e) => report.errors.push((i + 2, e.to_string())),
4043 }
4044 }
4045 report
4046}
4047
4048#[cfg(test)]
4049mod tests {
4050 use super::form_str_to_sea_value;
4051 use crate::migrate::Column;
4052 use crate::orm::{FkAction, SqlType};
4053 use sea_query::Value as SeaValue;
4054
4055 fn col(name: &str, ty: SqlType, nullable: bool) -> Column {
4056 Column {
4057 name: name.to_string(),
4058 ty,
4059 primary_key: false,
4060 nullable,
4061 fk_target: None,
4062 noform: false,
4063 privileged: false,
4064 db_constraint: true,
4065 noedit: false,
4066 auto_user_add: false,
4067 auto_user: false,
4068 is_string_repr: false,
4069 max_length: 0,
4070 choices: Vec::new(),
4071 choice_labels: Vec::new(),
4072 default: String::new(),
4073 is_multichoice: false,
4074 unique: false,
4075 on_delete: FkAction::NoAction,
4076 on_update: FkAction::NoAction,
4077 index: false,
4078 auto_now_add: false,
4079 auto_now: false,
4080 trim: false,
4081 lowercase: false,
4082 case_insensitive: false,
4083 help: String::new(),
4084 example: String::new(),
4085 widget: None,
4086 supported_backends: Vec::new(),
4087 min: None,
4088 max: None,
4089 text_format: None,
4090 slug_from: None,
4091 }
4092 }
4093
4094 #[test]
4095 fn form_fk_numeric_string_binds_as_bigint() {
4096 let mut plugin = col("plugin", SqlType::ForeignKey, false);
4097 plugin.fk_target = Some("plugin".to_string());
4098
4099 let value = form_str_to_sea_value(&plugin, "1").expect("coerce FK id");
4100
4101 assert_eq!(
4102 value,
4103 SeaValue::BigInt(Some(1)),
4104 "integer-backed FK form values must bind as bigint, not text"
4105 );
4106 }
4107
4108 #[test]
4109 fn nullable_form_fk_blank_binds_as_null_bigint() {
4110 let mut parent = col("parent", SqlType::ForeignKey, true);
4111 parent.fk_target = Some("plugin_comment".to_string());
4112
4113 let value = form_str_to_sea_value(&parent, "").expect("blank nullable FK");
4114
4115 assert_eq!(
4116 value,
4117 SeaValue::BigInt(None),
4118 "blank nullable integer-backed FK should bind SQL NULL"
4119 );
4120 }
4121}