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}
149
150impl<'a> DynQuerySet<'a> {
151 pub fn for_meta(meta: &'a ModelMeta) -> Self {
155 let select_cols = meta.fields.iter().map(|c| c.name.clone()).collect();
156 Self {
157 meta,
158 where_clauses: Vec::new(),
159 order: Vec::new(),
160 limit: None,
161 offset: None,
162 select_cols,
163 with_deleted: false,
164 only_deleted: false,
165 hard_delete: false,
166 select_related: Vec::new(),
167 }
168 }
169
170 pub fn with_deleted(mut self) -> Self {
173 self.with_deleted = true;
174 self
175 }
176
177 pub fn only_deleted(mut self) -> Self {
180 self.only_deleted = true;
181 self
182 }
183
184 pub fn hard_delete(mut self) -> Self {
186 self.hard_delete = true;
187 self
188 }
189
190 fn effective_where_clauses(&self) -> Vec<Condition> {
191 let mut clauses = self.where_clauses.clone();
192 if self.meta.soft_delete {
193 if self.only_deleted {
194 clauses
195 .push(Condition::all().add(Expr::col(Alias::new("deleted_at")).is_not_null()));
196 } else if !self.with_deleted {
197 clauses.push(Condition::all().add(Expr::col(Alias::new("deleted_at")).is_null()));
198 }
199 }
200 clauses
201 }
202
203 fn live_where_clauses(&self) -> Vec<Condition> {
204 let mut clauses = self.where_clauses.clone();
205 if self.meta.soft_delete {
206 clauses.push(Condition::all().add(Expr::col(Alias::new("deleted_at")).is_null()));
207 }
208 clauses
209 }
210
211 pub fn select_cols(mut self, cols: &[String]) -> Self {
215 let valid: Vec<String> = cols
216 .iter()
217 .filter(|n| self.meta.fields.iter().any(|c| &c.name == *n))
218 .cloned()
219 .collect();
220 if !valid.is_empty() {
221 self.select_cols = valid;
222 }
223 self
224 }
225
226 pub fn select_related_dyn(mut self, fields: &[String]) -> Self {
253 for name in fields {
254 let canonical = normalize_sr_token(name);
255 if validate_sr_chain(self.meta, &canonical).is_none() {
256 continue;
257 }
258 if !self.select_related.iter().any(|n| n == &canonical) {
259 self.select_related.push(canonical);
260 }
261 }
262 self
263 }
264
265 #[doc(hidden)]
268 pub fn select_related_fields(&self) -> &[String] {
269 &self.select_related
270 }
271
272 pub fn search(mut self, fields: &[String], term: &str) -> Self {
297 let term = term.trim();
298 if term.is_empty() {
299 return self;
300 }
301
302 let restricted = !fields.is_empty();
303 let as_int = term.parse::<i64>().ok();
304 let as_float = term.parse::<f64>().ok();
305 let as_bool = match term.to_ascii_lowercase().as_str() {
306 "true" => Some(true),
307 "false" => Some(false),
308 _ => None,
309 };
310 let like_pat = format!("%{}%", crate::orm::escape_like_literal(term)).to_uppercase();
313
314 let mut cond = Condition::any();
315 let mut added = 0;
316 for col in &self.meta.fields {
317 if restricted && !fields.iter().any(|f| f == &col.name) {
318 continue;
319 }
320 let predicate: Option<sea_query::SimpleExpr> = match col.ty {
321 SqlType::Text => Some(
322 Expr::expr(Func::upper(Expr::col(Alias::new(&col.name))))
323 .like(sea_query::LikeExpr::new(like_pat.clone()).escape('\\')),
324 ),
325 SqlType::SmallInt | SqlType::Integer | SqlType::BigInt | SqlType::ForeignKey => {
326 as_int.map(|n| Expr::col(Alias::new(&col.name)).eq(n))
327 }
328 SqlType::Real | SqlType::Double => {
329 as_float.map(|n| Expr::col(Alias::new(&col.name)).eq(n))
330 }
331 SqlType::Boolean => as_bool.map(|b| Expr::col(Alias::new(&col.name)).eq(b)),
332 _ => None,
333 };
334 if let Some(p) = predicate {
335 cond = cond.add(p);
336 added += 1;
337 }
338 }
339 if added > 0 {
340 self.where_clauses.push(cond);
341 }
342 self
343 }
344
345 pub fn filter_condition(mut self, cond: sea_query::Condition) -> Self {
351 self.where_clauses.push(cond);
352 self
353 }
354
355 pub fn filter_in_i64(mut self, col: &str, vals: &[i64]) -> Self {
358 if vals.is_empty() || !self.meta.fields.iter().any(|c| c.name == col) {
359 return self;
360 }
361 let cond = Condition::all().add(Expr::col(Alias::new(col)).is_in(vals.iter().copied()));
362 self.where_clauses.push(cond);
363 self
364 }
365
366 pub fn filter_m2m_contains_any(mut self, field_name: &str, child_ids: &[String]) -> Self {
390 if child_ids.is_empty() {
391 return self;
392 }
393 let Some(rel) = self
394 .meta
395 .m2m_relations
396 .iter()
397 .find(|r| r.field_name == field_name)
398 else {
399 return self;
400 };
401 let Some(pk_col) = self.meta.pk_column() else {
402 return self;
403 };
404 let target_pk_ty = crate::migrate::pk_meta_for_table(&rel.target_table)
415 .map(|(_, ty)| ty)
416 .unwrap_or(SqlType::BigInt);
417 let junction_table = format!("{}_{}", self.meta.table, rel.field_name);
418 let child_id_expr = Expr::col(Alias::new("child_id"));
419 let in_clause: sea_query::SimpleExpr = match target_pk_ty {
420 SqlType::Text | SqlType::Uuid => {
421 let bound: Vec<String> = child_ids
425 .iter()
426 .filter_map(|s| {
427 let s = s.trim();
428 if s.is_empty() {
429 None
430 } else {
431 Some(s.to_string())
432 }
433 })
434 .collect();
435 if bound.is_empty() {
436 return self;
437 }
438 child_id_expr.is_in(bound)
439 }
440 _ => {
441 let parsed: Vec<i64> = child_ids.iter().filter_map(|s| s.parse().ok()).collect();
445 if parsed.is_empty() {
446 return self;
447 }
448 child_id_expr.is_in(parsed)
449 }
450 };
451 let subq = Query::select()
452 .column(Alias::new("parent_id"))
453 .from(crate::db::router::schema_qualified_table(&junction_table))
454 .and_where(in_clause)
455 .to_owned();
456 let cond =
457 Condition::all().add(Expr::col(Alias::new(pk_col.name.clone())).in_subquery(subq));
458 self.where_clauses.push(cond);
459 self
460 }
461
462 pub fn filter_in_strings(mut self, col: &str, vals: &[String]) -> Self {
474 let Some(meta_col) = self.meta.fields.iter().find(|c| c.name == col) else {
475 return self;
476 };
477 if vals.is_empty() {
478 return self;
479 }
480 let expr = Expr::col(Alias::new(col));
481 let cond = match crate::migrate::fk_effective_type(meta_col) {
488 SqlType::SmallInt | SqlType::Integer => {
489 let parsed: Vec<i32> = vals.iter().filter_map(|s| s.parse().ok()).collect();
490 if parsed.is_empty() {
491 return self;
492 }
493 Condition::all().add(expr.is_in(parsed))
494 }
495 SqlType::BigInt | SqlType::ForeignKey => {
496 let parsed: Vec<i64> = vals.iter().filter_map(|s| s.parse().ok()).collect();
497 if parsed.is_empty() {
498 return self;
499 }
500 Condition::all().add(expr.is_in(parsed))
501 }
502 SqlType::Real | SqlType::Double => {
503 let parsed: Vec<f64> = vals.iter().filter_map(|s| s.parse().ok()).collect();
504 if parsed.is_empty() {
505 return self;
506 }
507 Condition::all().add(expr.is_in(parsed))
508 }
509 SqlType::Boolean => {
510 let parsed: Vec<bool> = vals
511 .iter()
512 .map(|s| matches!(s.as_str(), "true" | "on" | "1"))
513 .collect();
514 Condition::all().add(expr.is_in(parsed))
515 }
516 SqlType::Uuid => {
521 let parsed: Vec<uuid::Uuid> = vals
522 .iter()
523 .filter_map(|s| uuid::Uuid::parse_str(s).ok())
524 .collect();
525 if parsed.is_empty() {
526 return self;
527 }
528 Condition::all().add(expr.is_in(parsed))
529 }
530 _ => Condition::all().add(expr.is_in(vals.iter().map(|s| s.to_string()))),
531 };
532 self.where_clauses.push(cond);
533 self
534 }
535
536 pub fn filter_eq_string(mut self, col: &str, value: &str) -> Self {
540 let Some(meta_col) = self.meta.fields.iter().find(|c| c.name == col) else {
541 return self;
542 };
543 let expr = Expr::col(Alias::new(col));
544 let predicate = match crate::migrate::fk_effective_type(meta_col) {
547 SqlType::SmallInt | SqlType::Integer => value.parse::<i32>().ok().map(|v| expr.eq(v)),
548 SqlType::BigInt | SqlType::ForeignKey => value.parse::<i64>().ok().map(|v| expr.eq(v)),
549 SqlType::Real | SqlType::Double => value.parse::<f64>().ok().map(|v| expr.eq(v)),
550 SqlType::Boolean => {
551 let v = matches!(value, "true" | "on" | "1");
552 Some(expr.eq(v))
553 }
554 SqlType::Uuid => uuid::Uuid::parse_str(value).ok().map(|u| expr.eq(u)),
557 _ => Some(expr.eq(value.to_string())),
558 };
559 if let Some(p) = predicate {
560 self.where_clauses.push(Condition::all().add(p));
561 }
562 self
563 }
564
565 pub fn order_by_col(mut self, col: &str, descending: bool) -> Self {
568 if self.meta.fields.iter().any(|c| c.name == col) {
569 self.order.push((col.to_string(), descending));
570 }
571 self
572 }
573
574 pub fn limit(mut self, n: u64) -> Self {
576 self.limit = Some(n);
577 self
578 }
579
580 pub fn offset(mut self, n: u64) -> Self {
582 self.offset = Some(n);
583 self
584 }
585
586 pub async fn count(self) -> Result<i64, DynError> {
590 let mut q = Query::select();
591 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
592 q.expr(Func::count(Expr::col(Asterisk)));
593 let where_clauses = self.effective_where_clauses();
594 for cond in &where_clauses {
595 q.cond_where(cond.clone());
596 }
597
598 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Read) {
599 DbPool::Sqlite(pool) => {
600 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
601 let row = sqlx::query_with(&sql, values).fetch_one(&pool).await?;
602 Ok(row.try_get::<i64, _>(0)?)
603 }
604 DbPool::Postgres(pool) => {
605 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
606 let row = sqlx::query_with(&sql, values).fetch_one(&pool).await?;
607 Ok(row.try_get::<i64, _>(0)?)
608 }
609 }
610 }
611
612 pub async fn fetch_distinct_strings(self, col: &str) -> Result<Vec<String>, DynError> {
617 let Some(col_meta) = self.meta.fields.iter().find(|c| c.name == col) else {
618 return Ok(Vec::new());
619 };
620 let mut q = Query::select();
621 q.distinct();
622 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
623 q.column(Alias::new(col));
624 let where_clauses = self.effective_where_clauses();
625 for cond in &where_clauses {
626 q.cond_where(cond.clone());
627 }
628 if let Some(n) = self.limit {
629 q.limit(n);
630 }
631
632 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Read) {
633 DbPool::Sqlite(pool) => {
634 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
635 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
636 let mut out = Vec::with_capacity(rows.len());
637 for row in rows {
638 out.push(decode_to_string(&row, col_meta)?);
639 }
640 Ok(out)
641 }
642 DbPool::Postgres(pool) => {
643 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
644 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
645 let mut out = Vec::with_capacity(rows.len());
646 for row in rows {
647 out.push(decode_pg_to_string(&row, col_meta)?);
648 }
649 Ok(out)
650 }
651 }
652 }
653
654 pub async fn delete(self) -> Result<u64, DynError> {
663 if self.meta.soft_delete && !self.hard_delete {
664 return self.soft_delete_update().await;
665 }
666 let where_clauses = self.effective_where_clauses();
667 let parent_pks: Vec<serde_json::Value> = match self.meta.pk_column() {
671 Some(pk_col) => collect_parent_pks(&self.meta, pk_col, &where_clauses)
672 .await
673 .unwrap_or_default(),
674 None => Vec::new(),
675 };
676
677 let mut q = Query::delete();
678 q.from_table(crate::db::router::schema_qualified_table(&self.meta.table));
679 for cond in &where_clauses {
680 q.cond_where(cond.clone());
681 }
682
683 let rows_affected = match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
684 DbPool::Sqlite(pool) => {
685 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
686 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
687 res.rows_affected()
688 }
689 DbPool::Postgres(pool) => {
690 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
691 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
692 res.rows_affected()
693 }
694 };
695
696 crate::signals::emit_bulk_post_delete_by_table(&self.meta.table, parent_pks).await;
701 Ok(rows_affected)
702 }
703
704 async fn soft_delete_update(self) -> Result<u64, DynError> {
705 let where_clauses = self.live_where_clauses();
706 let parent_pks: Vec<serde_json::Value> = match self.meta.pk_column() {
707 Some(pk_col) => collect_parent_pks(self.meta, pk_col, &where_clauses)
708 .await
709 .unwrap_or_default(),
710 None => Vec::new(),
711 };
712
713 let mut q = Query::update();
714 q.table(crate::db::router::schema_qualified_table(&self.meta.table));
715 q.value(
716 Alias::new("deleted_at"),
717 sea_query::Value::ChronoDateTimeUtc(Some(Box::new(chrono::Utc::now()))),
718 );
719 for cond in &where_clauses {
720 q.cond_where(cond.clone());
721 }
722
723 let rows_affected = match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
724 DbPool::Sqlite(pool) => {
725 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
726 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
727 res.rows_affected()
728 }
729 DbPool::Postgres(pool) => {
730 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
731 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
732 res.rows_affected()
733 }
734 };
735
736 crate::signals::emit_bulk_post_delete_by_table(&self.meta.table, parent_pks).await;
737 Ok(rows_affected)
738 }
739
740 pub async fn restore(self) -> Result<u64, DynError> {
751 if !self.meta.soft_delete {
752 return Ok(0);
753 }
754 let mut where_clauses = self.where_clauses.clone();
758 where_clauses
759 .push(Condition::all().add(Expr::col(Alias::new("deleted_at")).is_not_null()));
760
761 let parent_pks: Vec<serde_json::Value> = match self.meta.pk_column() {
762 Some(pk_col) => collect_parent_pks(self.meta, pk_col, &where_clauses)
763 .await
764 .unwrap_or_default(),
765 None => Vec::new(),
766 };
767
768 let mut q = Query::update();
769 q.table(crate::db::router::schema_qualified_table(&self.meta.table));
770 q.value(
771 Alias::new("deleted_at"),
772 sea_query::Value::ChronoDateTimeUtc(None),
773 );
774 for cond in &where_clauses {
775 q.cond_where(cond.clone());
776 }
777
778 let rows_affected = match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
779 DbPool::Sqlite(pool) => {
780 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
781 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
782 res.rows_affected()
783 }
784 DbPool::Postgres(pool) => {
785 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
786 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
787 res.rows_affected()
788 }
789 };
790
791 crate::signals::emit_bulk_post_save_by_table(&self.meta.table, parent_pks, false).await;
795 Ok(rows_affected)
796 }
797
798 pub async fn update_one(self, col: &str, value: &str) -> Result<u64, DynError> {
803 let Some(col_meta) = self.meta.fields.iter().find(|c| c.name == col) else {
804 return Ok(0);
805 };
806 let sea_value = match form_str_to_sea_value(col_meta, value) {
807 Ok(v) => v,
808 Err(e) => {
810 return Err(DynError::Write(WriteError::Validator {
811 field: col_meta.name.clone(),
812 message: e.to_string(),
813 }));
814 }
815 };
816
817 let mut q = Query::update();
818 q.table(crate::db::router::schema_qualified_table(&self.meta.table));
819 q.value(Alias::new(col), sea_value);
820 let where_clauses = self.effective_where_clauses();
821 for cond in &where_clauses {
822 q.cond_where(cond.clone());
823 }
824
825 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
826 DbPool::Sqlite(pool) => {
827 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
828 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
829 Ok(res.rows_affected())
830 }
831 DbPool::Postgres(pool) => {
832 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
833 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
834 Ok(res.rows_affected())
835 }
836 }
837 }
838
839 pub async fn update_form(
846 self,
847 form: &HashMap<String, String>,
848 skip: &[String],
849 ) -> Result<u64, DynError> {
850 let Some(q) = self.build_update_form_query(form, skip)? else {
851 return Ok(0);
852 };
853
854 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
855 DbPool::Sqlite(pool) => {
856 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
857 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
858 Ok(res.rows_affected())
859 }
860 DbPool::Postgres(pool) => {
861 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
862 let res = sqlx::query_with(&sql, values).execute(&pool).await?;
863 Ok(res.rows_affected())
864 }
865 }
866 }
867
868 fn build_update_form_query(
876 &self,
877 form: &HashMap<String, String>,
878 skip: &[String],
879 ) -> Result<Option<sea_query::UpdateStatement>, DynError> {
880 let mut q = Query::update();
881 q.table(crate::db::router::schema_qualified_table(&self.meta.table));
882 let mut any = false;
883 for col in &self.meta.fields {
884 if col.primary_key || skip.iter().any(|s| s == &col.name) {
885 continue;
886 }
887 if col.auto_now {
895 q.value(
896 Alias::new(&col.name),
897 crate::orm::write::now_for_column(col.ty),
898 );
899 any = true;
900 continue;
901 }
902 let Some(raw) = form.get(&col.name) else {
903 continue;
904 };
905 let sea_value = match form_str_to_sea_value(col, raw) {
906 Ok(v) => v,
907 Err(e) => {
913 return Err(DynError::Write(WriteError::Validator {
914 field: col.name.clone(),
915 message: e.to_string(),
916 }));
917 }
918 };
919 q.value(Alias::new(&col.name), sea_value);
920 any = true;
921 }
922 if !any {
923 return Ok(None);
924 }
925 let where_clauses = self.effective_where_clauses();
926 for cond in &where_clauses {
927 q.cond_where(cond.clone());
928 }
929 Ok(Some(q))
930 }
931
932 pub async fn update_form_in_tx(
940 self,
941 tx: &mut crate::db::Transaction,
942 form: &HashMap<String, String>,
943 skip: &[String],
944 ) -> Result<u64, DynError> {
945 let Some(q) = self.build_update_form_query(form, skip)? else {
946 return Ok(0);
947 };
948
949 match tx.backend_name() {
950 "sqlite" => {
951 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
952 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
953 let res = sqlx::query_with(&sql, values).execute(&mut **inner).await?;
954 Ok(res.rows_affected())
955 }
956 _ => {
957 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
958 let inner = tx.as_pg_mut().expect("postgres backend_name");
959 let res = sqlx::query_with(&sql, values).execute(&mut **inner).await?;
960 Ok(res.rows_affected())
961 }
962 }
963 }
964
965 pub async fn insert_form(
971 self,
972 form: &HashMap<String, String>,
973 skip: &[String],
974 ) -> Result<i64, DynError> {
975 let Some(mut q) = self.build_insert_form_query(form, skip)? else {
976 return Ok(0);
977 };
978
979 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
980 DbPool::Sqlite(pool) => {
981 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
982 let res = sqlx::query_with(&sql, vals).execute(&pool).await?;
983 Ok(res.last_insert_rowid())
984 }
985 DbPool::Postgres(pool) => {
986 let pk_name = self
992 .meta
993 .fields
994 .iter()
995 .find(|c| c.primary_key)
996 .map(|c| c.name.clone());
997 if let Some(pk) = pk_name {
998 q.returning_col(Alias::new(&pk));
999 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1000 let row = sqlx::query_with(&sql, vals).fetch_one(&pool).await?;
1001 Ok(row.try_get::<i64, _>(pk.as_str()).unwrap_or(0))
1002 } else {
1003 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1004 let _ = sqlx::query_with(&sql, vals).execute(&pool).await?;
1005 Ok(0)
1006 }
1007 }
1008 }
1009 }
1010
1011 fn build_insert_form_query(
1020 &self,
1021 form: &HashMap<String, String>,
1022 skip: &[String],
1023 ) -> Result<Option<sea_query::InsertStatement>, DynError> {
1024 let mut cols: Vec<&str> = Vec::new();
1025 let mut values: Vec<SeaValue> = Vec::new();
1026 for col in &self.meta.fields {
1027 if skip.iter().any(|s| s == &col.name) {
1028 continue;
1029 }
1030 if col.primary_key
1033 && matches!(
1034 col.ty,
1035 SqlType::Integer | SqlType::BigInt | SqlType::SmallInt
1036 )
1037 && form.get(&col.name).is_none_or(|v| v.is_empty())
1038 {
1039 continue;
1040 }
1041 if (col.auto_now_add || col.auto_now)
1049 && form.get(&col.name).is_none_or(|v| v.is_empty())
1050 {
1051 cols.push(&col.name);
1052 values.push(crate::orm::write::now_for_column(col.ty));
1053 continue;
1054 }
1055 let raw = form.get(&col.name).map(|s| s.as_str()).unwrap_or("");
1056 let sea_value = match form_str_to_sea_value(col, raw) {
1057 Ok(v) => v,
1058 Err(e) => {
1061 return Err(DynError::Write(WriteError::Validator {
1062 field: col.name.clone(),
1063 message: e.to_string(),
1064 }));
1065 }
1066 };
1067 cols.push(&col.name);
1068 values.push(sea_value);
1069 }
1070 if cols.is_empty() {
1071 return Ok(None);
1072 }
1073
1074 let mut q = Query::insert();
1075 q.into_table(crate::db::router::schema_qualified_table(&self.meta.table));
1076 q.columns(cols.iter().map(|c| Alias::new(*c)).collect::<Vec<_>>());
1077 let exprs: Vec<sea_query::SimpleExpr> = values.into_iter().map(Into::into).collect();
1078 q.values_panic(exprs);
1079 Ok(Some(q))
1080 }
1081
1082 pub async fn insert_form_in_tx(
1091 self,
1092 tx: &mut crate::db::Transaction,
1093 form: &HashMap<String, String>,
1094 skip: &[String],
1095 ) -> Result<i64, DynError> {
1096 let Some(mut q) = self.build_insert_form_query(form, skip)? else {
1097 return Ok(0);
1098 };
1099
1100 match tx.backend_name() {
1101 "sqlite" => {
1102 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
1103 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1104 let res = sqlx::query_with(&sql, vals).execute(&mut **inner).await?;
1105 Ok(res.last_insert_rowid())
1106 }
1107 _ => {
1108 let pk_name = self
1112 .meta
1113 .fields
1114 .iter()
1115 .find(|c| c.primary_key)
1116 .map(|c| c.name.clone());
1117 let inner = tx.as_pg_mut().expect("postgres backend_name");
1118 if let Some(pk) = pk_name {
1119 q.returning_col(Alias::new(&pk));
1120 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1121 let row = sqlx::query_with(&sql, vals).fetch_one(&mut **inner).await?;
1122 Ok(row.try_get::<i64, _>(pk.as_str()).unwrap_or(0))
1123 } else {
1124 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1125 let _ = sqlx::query_with(&sql, vals).execute(&mut **inner).await?;
1126 Ok(0)
1127 }
1128 }
1129 }
1130 }
1131
1132 pub async fn fetch_as_strings(self) -> Result<Vec<HashMap<String, String>>, DynError> {
1137 let mut q = Query::select();
1138 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
1139 for c in &self.select_cols {
1140 q.column(Alias::new(c));
1141 }
1142 let where_clauses = self.effective_where_clauses();
1143 for cond in &where_clauses {
1144 q.cond_where(cond.clone());
1145 }
1146 for (col, descending) in &self.order {
1147 q.order_by(
1148 Alias::new(col),
1149 if *descending { Order::Desc } else { Order::Asc },
1150 );
1151 }
1152 if let Some(n) = self.limit {
1153 q.limit(n);
1154 }
1155 if let Some(n) = self.offset {
1156 q.offset(n);
1157 }
1158
1159 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Read) {
1160 DbPool::Sqlite(pool) => {
1161 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1162 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
1163 let mut out: Vec<HashMap<String, String>> = Vec::with_capacity(rows.len());
1164 for row in rows {
1165 let mut entry = HashMap::new();
1166 for col_name in &self.select_cols {
1167 if let Some(col_meta) =
1168 self.meta.fields.iter().find(|c| &c.name == col_name)
1169 {
1170 let v = decode_to_string(&row, col_meta)?;
1171 entry.insert(col_name.clone(), v);
1172 }
1173 }
1174 out.push(entry);
1175 }
1176 Ok(out)
1177 }
1178 DbPool::Postgres(pool) => {
1179 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1180 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
1181 let mut out: Vec<HashMap<String, String>> = Vec::with_capacity(rows.len());
1182 for row in rows {
1183 let mut entry = HashMap::new();
1184 for col_name in &self.select_cols {
1185 if let Some(col_meta) =
1186 self.meta.fields.iter().find(|c| &c.name == col_name)
1187 {
1188 let v = decode_pg_to_string(&row, col_meta)?;
1189 entry.insert(col_name.clone(), v);
1190 }
1191 }
1192 out.push(entry);
1193 }
1194 Ok(out)
1195 }
1196 }
1197 }
1198
1199 pub async fn fetch_as_json(
1205 self,
1206 ) -> Result<Vec<serde_json::Map<String, serde_json::Value>>, DynError> {
1207 let mut q = Query::select();
1208 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
1209 for c in &self.select_cols {
1210 q.column(Alias::new(c));
1211 }
1212 let where_clauses = self.effective_where_clauses();
1213 for cond in &where_clauses {
1214 q.cond_where(cond.clone());
1215 }
1216 for (col, descending) in &self.order {
1217 q.order_by(
1218 Alias::new(col),
1219 if *descending { Order::Desc } else { Order::Asc },
1220 );
1221 }
1222 if let Some(n) = self.limit {
1223 q.limit(n);
1224 }
1225 if let Some(n) = self.offset {
1226 q.offset(n);
1227 }
1228
1229 let pk_name = self
1230 .meta
1231 .pk_column()
1232 .map(|c| c.name.clone())
1233 .unwrap_or_default();
1234 let selected_cols: Vec<(&String, &Column)> = self
1235 .select_cols
1236 .iter()
1237 .filter_map(|col_name| {
1238 self.meta
1239 .fields
1240 .iter()
1241 .find(|c| &c.name == col_name)
1242 .map(|col| (col_name, col))
1243 })
1244 .collect();
1245 let mut out: Vec<serde_json::Map<String, serde_json::Value>> =
1246 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Read) {
1247 DbPool::Sqlite(pool) => {
1248 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1249 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
1250 let mut out: Vec<serde_json::Map<String, serde_json::Value>> =
1251 Vec::with_capacity(rows.len());
1252 for row in rows {
1253 let mut entry = serde_json::Map::new();
1254 for (col_name, col_meta) in &selected_cols {
1255 entry.insert((*col_name).clone(), decode_to_json(&row, col_meta)?);
1256 }
1257 out.push(entry);
1258 }
1259 out
1260 }
1261 DbPool::Postgres(pool) => {
1262 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1263 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
1264 let mut out: Vec<serde_json::Map<String, serde_json::Value>> =
1265 Vec::with_capacity(rows.len());
1266 for row in rows {
1267 let mut entry = serde_json::Map::new();
1268 for (col_name, col_meta) in &selected_cols {
1269 entry.insert((*col_name).clone(), decode_pg_to_json(&row, col_meta)?);
1270 }
1271 out.push(entry);
1272 }
1273 out
1274 }
1275 };
1276
1277 if !self.meta.m2m_relations.is_empty() && !out.is_empty() {
1287 hydrate_m2m_batched(&self.meta, &pk_name, &mut out).await?;
1288 }
1289
1290 if !self.select_related.is_empty() && !out.is_empty() {
1299 hydrate_select_related_into(&self.meta, &self.select_related, &mut out).await?;
1300 }
1301 Ok(out)
1302 }
1303
1304 pub async fn first_as_json(
1307 mut self,
1308 ) -> Result<Option<serde_json::Map<String, serde_json::Value>>, DynError> {
1309 self.limit = Some(1);
1310 let mut rows = self.fetch_as_json().await?;
1311 Ok(rows.pop())
1312 }
1313
1314 pub async fn fetch_one_json_in_tx(
1324 self,
1325 tx: &mut crate::db::Transaction,
1326 ) -> Result<Option<serde_json::Map<String, serde_json::Value>>, DynError> {
1327 let mut q = Query::select();
1328 q.from(crate::db::router::schema_qualified_table(&self.meta.table));
1329 for c in &self.meta.fields {
1330 q.column(Alias::new(&c.name));
1331 }
1332 let where_clauses = self.effective_where_clauses();
1333 for cond in &where_clauses {
1334 q.cond_where(cond.clone());
1335 }
1336 q.limit(1);
1337
1338 let out = match tx.backend_name() {
1339 "sqlite" => {
1340 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1341 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1342 let row = sqlx::query_with(&sql, values)
1343 .fetch_optional(&mut **inner)
1344 .await?;
1345 match row {
1346 Some(row) => {
1347 let mut entry = serde_json::Map::new();
1348 for col in &self.meta.fields {
1349 entry.insert(col.name.clone(), decode_to_json(&row, col)?);
1350 }
1351 Some(entry)
1352 }
1353 None => None,
1354 }
1355 }
1356 _ => {
1357 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1358 let inner = tx.as_pg_mut().expect("postgres backend_name");
1359 let row = sqlx::query_with(&sql, values)
1360 .fetch_optional(&mut **inner)
1361 .await?;
1362 match row {
1363 Some(row) => {
1364 let mut entry = serde_json::Map::new();
1365 for col in &self.meta.fields {
1366 entry.insert(col.name.clone(), decode_pg_to_json(&row, col)?);
1367 }
1368 Some(entry)
1369 }
1370 None => None,
1371 }
1372 }
1373 };
1374 Ok(out)
1375 }
1376
1377 pub async fn insert_json(
1385 self,
1386 body: &serde_json::Map<String, serde_json::Value>,
1387 ) -> Result<serde_json::Map<String, serde_json::Value>, crate::orm::write::WriteError> {
1388 use crate::orm::write::WriteError;
1389
1390 let body_owned: serde_json::Map<String, serde_json::Value>;
1393 let body: &serde_json::Map<String, serde_json::Value> =
1394 match normalise_insert_body(self.meta, body) {
1395 Some(owned) => {
1396 body_owned = owned;
1397 &body_owned
1398 }
1399 None => body,
1400 };
1401
1402 let validation_errors = crate::orm::validation::validate_on_create(self.meta, body).await;
1404 if !validation_errors.is_empty() {
1405 return Err(WriteError::Multiple {
1406 errors: validation_errors,
1407 });
1408 }
1409
1410 let InsertPlan {
1413 mut q,
1414 pk_name,
1415 pk_ty,
1416 } = build_insert_plan(self.meta, body)?;
1417
1418 crate::signals::emit_pre_save_by_table(
1424 &self.meta.table,
1425 serde_json::Value::Object(body.clone()),
1426 true,
1427 )
1428 .await;
1429
1430 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
1431 DbPool::Sqlite(pool) => {
1432 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
1433 let res = sqlx::query_with(&sql, vals)
1434 .execute(&pool)
1435 .await
1436 .map_err(|e| classify_or_sqlx(e, body))?;
1437 let pk_pred = match pk_ty {
1441 SqlType::Integer | SqlType::BigInt | SqlType::SmallInt => {
1442 Expr::col(Alias::new(&pk_name)).eq(res.last_insert_rowid())
1443 }
1444 _ => {
1445 let supplied = body
1448 .get(&pk_name)
1449 .cloned()
1450 .unwrap_or(serde_json::Value::Null);
1451 let sea_value = crate::orm::write::json_to_sea_value(
1452 pk_ty, &supplied, false, &pk_name, None,
1453 )?;
1454 Expr::col(Alias::new(&pk_name)).eq(sea_value)
1455 }
1456 };
1457 let mut sel = Query::select();
1458 sel.from(crate::db::router::schema_qualified_table(&self.meta.table));
1459 for c in &self.meta.fields {
1460 sel.column(Alias::new(&c.name));
1461 }
1462 sel.cond_where(Condition::all().add(pk_pred));
1463 let (sel_sql, sel_vals) = sel.build_sqlx(SqliteQueryBuilder);
1464 let row = sqlx::query_with(&sel_sql, sel_vals)
1465 .fetch_one(&pool)
1466 .await?;
1467 let mut out = serde_json::Map::new();
1468 for col in &self.meta.fields {
1469 out.insert(col.name.clone(), decode_to_json(&row, col)?);
1470 }
1471 let pk_value = out.get(&pk_name).cloned();
1477 write_m2m_junctions(&self.meta, pk_value.as_ref(), body).await?;
1478 hydrate_m2m_into(&self.meta, pk_value.as_ref(), &mut out).await?;
1482 crate::signals::emit_post_save_by_table(
1484 &self.meta.table,
1485 serde_json::Value::Object(out.clone()),
1486 true,
1487 )
1488 .await;
1489 Ok(out)
1490 }
1491 DbPool::Postgres(pool) => {
1492 q.returning_all();
1497 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1498 let row = sqlx::query_with(&sql, vals)
1499 .fetch_one(&pool)
1500 .await
1501 .map_err(|e| classify_or_sqlx(e, body))?;
1502 let mut out = serde_json::Map::new();
1503 for col in &self.meta.fields {
1504 out.insert(col.name.clone(), decode_pg_to_json(&row, col)?);
1505 }
1506 let pk_value = out.get(&pk_name).cloned();
1507 write_m2m_junctions(&self.meta, pk_value.as_ref(), body).await?;
1508 hydrate_m2m_into(&self.meta, pk_value.as_ref(), &mut out).await?;
1509 crate::signals::emit_post_save_by_table(
1511 &self.meta.table,
1512 serde_json::Value::Object(out.clone()),
1513 true,
1514 )
1515 .await;
1516 Ok(out)
1517 }
1518 }
1519 }
1520
1521 pub async fn insert_json_in_tx(
1545 self,
1546 body: &serde_json::Map<String, serde_json::Value>,
1547 tx: &mut crate::db::Transaction,
1548 ) -> Result<serde_json::Map<String, serde_json::Value>, crate::orm::write::WriteError> {
1549 use crate::orm::write::WriteError;
1550
1551 let body_owned: serde_json::Map<String, serde_json::Value>;
1553 let body: &serde_json::Map<String, serde_json::Value> =
1554 match normalise_insert_body(self.meta, body) {
1555 Some(owned) => {
1556 body_owned = owned;
1557 &body_owned
1558 }
1559 None => body,
1560 };
1561
1562 let validation_errors =
1565 crate::orm::validation::validate_on_create_in_tx(self.meta, body, tx).await;
1566 if !validation_errors.is_empty() {
1567 return Err(WriteError::Multiple {
1568 errors: validation_errors,
1569 });
1570 }
1571
1572 let InsertPlan {
1574 mut q,
1575 pk_name,
1576 pk_ty,
1577 } = build_insert_plan(self.meta, body)?;
1578
1579 match tx.backend_name() {
1580 "sqlite" => {
1581 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
1582 let res = {
1583 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1584 sqlx::query_with(&sql, vals)
1585 .execute(&mut **inner)
1586 .await
1587 .map_err(|e| classify_or_sqlx(e, body))?
1588 };
1589 let pk_pred = match pk_ty {
1592 SqlType::Integer | SqlType::BigInt | SqlType::SmallInt => {
1593 Expr::col(Alias::new(&pk_name)).eq(res.last_insert_rowid())
1594 }
1595 _ => {
1596 let supplied = body
1597 .get(&pk_name)
1598 .cloned()
1599 .unwrap_or(serde_json::Value::Null);
1600 let sea_value = crate::orm::write::json_to_sea_value(
1601 pk_ty, &supplied, false, &pk_name, None,
1602 )?;
1603 Expr::col(Alias::new(&pk_name)).eq(sea_value)
1604 }
1605 };
1606 let mut sel = Query::select();
1607 sel.from(crate::db::router::schema_qualified_table(&self.meta.table));
1608 for c in &self.meta.fields {
1609 sel.column(Alias::new(&c.name));
1610 }
1611 sel.cond_where(Condition::all().add(pk_pred));
1612 let (sel_sql, sel_vals) = sel.build_sqlx(SqliteQueryBuilder);
1613 let mut out = serde_json::Map::new();
1614 {
1615 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1616 let row = sqlx::query_with(&sel_sql, sel_vals)
1617 .fetch_one(&mut **inner)
1618 .await?;
1619 for col in &self.meta.fields {
1620 out.insert(col.name.clone(), decode_to_json(&row, col)?);
1621 }
1622 }
1623 let pk_value = out.get(&pk_name).cloned();
1625 write_m2m_junctions_in_tx(self.meta, pk_value.as_ref(), body, tx).await?;
1626 hydrate_m2m_into_tx(self.meta, pk_value.as_ref(), &mut out, tx).await?;
1627 Ok(out)
1628 }
1629 _ => {
1630 q.returning_all();
1631 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
1632 let mut out = serde_json::Map::new();
1633 {
1634 let inner = tx.as_pg_mut().expect("postgres backend_name");
1635 let row = sqlx::query_with(&sql, vals)
1636 .fetch_one(&mut **inner)
1637 .await
1638 .map_err(|e| classify_or_sqlx(e, body))?;
1639 for col in &self.meta.fields {
1640 out.insert(col.name.clone(), decode_pg_to_json(&row, col)?);
1641 }
1642 }
1643 let pk_value = out.get(&pk_name).cloned();
1644 write_m2m_junctions_in_tx(self.meta, pk_value.as_ref(), body, tx).await?;
1645 hydrate_m2m_into_tx(self.meta, pk_value.as_ref(), &mut out, tx).await?;
1646 Ok(out)
1647 }
1648 }
1649 }
1650
1651 pub async fn update_json_in_tx(
1662 self,
1663 body: &serde_json::Map<String, serde_json::Value>,
1664 tx: &mut crate::db::Transaction,
1665 ) -> Result<u64, crate::orm::write::WriteError> {
1666 use crate::orm::write::WriteError;
1667
1668 let needs_owned = self
1671 .meta
1672 .fields
1673 .iter()
1674 .any(|c| c.noform || c.slug_from.is_some());
1675 let mut body_owned: serde_json::Map<String, serde_json::Value>;
1676 let body: &serde_json::Map<String, serde_json::Value> = if needs_owned {
1677 body_owned = body.clone();
1678 for col in &self.meta.fields {
1679 if col.noform {
1680 body_owned.remove(&col.name);
1681 }
1682 }
1683 crate::orm::write::apply_slug_from(&self.meta.fields, &mut body_owned, true);
1684 &body_owned
1685 } else {
1686 body
1687 };
1688
1689 let validation_errors =
1693 crate::orm::validation::validate_on_update_in_tx(self.meta, body, tx).await;
1694 if !validation_errors.is_empty() {
1695 return Err(WriteError::Multiple {
1696 errors: validation_errors,
1697 });
1698 }
1699
1700 let mut q = Query::update();
1701 q.table(crate::db::router::schema_qualified_table(&self.meta.table));
1702 let mut any = false;
1703 for col in &self.meta.fields {
1704 if col.primary_key {
1705 continue;
1706 }
1707 let Some(json) = body.get(&col.name) else {
1708 if col.auto_now {
1709 let now_value = crate::orm::write::now_for_column(col.ty);
1710 q.value(Alias::new(&col.name), now_value);
1711 any = true;
1712 }
1713 continue;
1714 };
1715 validate_numeric_bounds(col, json)?;
1716 if let (Some(fmt), Some(s)) = (col.text_format.as_deref(), json.as_str()) {
1717 if let Err(e) = crate::orm::validators::validate_text_format(fmt, s) {
1718 return Err(WriteError::Validator {
1719 field: col.name.clone(),
1720 message: e.to_string(),
1721 });
1722 }
1723 }
1724 let sea_value = crate::orm::write::json_to_sea_value(
1725 col.ty,
1726 json,
1727 col.nullable,
1728 &col.name,
1729 fk_target_pk_sql_type(col),
1730 )?;
1731 q.value(Alias::new(&col.name), sea_value);
1732 any = true;
1733 }
1734 let touches_m2m = self
1735 .meta
1736 .m2m_relations
1737 .iter()
1738 .any(|r| body.contains_key(&r.field_name));
1739 if !any && !touches_m2m {
1740 return Ok(0);
1741 }
1742 let where_clauses = self.effective_where_clauses();
1743 for cond in &where_clauses {
1744 q.cond_where(cond.clone());
1745 }
1746
1747 let parent_pks: Vec<serde_json::Value> = match self.meta.pk_column() {
1751 Some(pk_col) => {
1752 collect_parent_pks_in_tx(self.meta, pk_col, &where_clauses, tx).await?
1753 }
1754 None => Vec::new(),
1755 };
1756
1757 if any {
1758 match tx.backend_name() {
1759 "sqlite" => {
1760 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1761 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1762 sqlx::query_with(&sql, values)
1763 .execute(&mut **inner)
1764 .await
1765 .map_err(|e| classify_or_sqlx(e, body))?;
1766 }
1767 _ => {
1768 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1769 let inner = tx.as_pg_mut().expect("postgres backend_name");
1770 sqlx::query_with(&sql, values)
1771 .execute(&mut **inner)
1772 .await
1773 .map_err(|e| classify_or_sqlx(e, body))?;
1774 }
1775 }
1776 }
1777 for pk in &parent_pks {
1778 write_m2m_junctions_in_tx(self.meta, Some(pk), body, tx).await?;
1779 }
1780 Ok(parent_pks.len().max(if any { 1 } else { 0 }) as u64)
1781 }
1782
1783 pub async fn delete_in_tx(self, tx: &mut crate::db::Transaction) -> Result<u64, DynError> {
1791 let soft = self.meta.soft_delete && !self.hard_delete;
1792 let where_clauses = if soft {
1793 self.live_where_clauses()
1794 } else {
1795 self.effective_where_clauses()
1796 };
1797
1798 let table = crate::db::router::schema_qualified_table(&self.meta.table);
1802 let build = |is_sqlite: bool| {
1803 if soft {
1804 let mut u = Query::update();
1805 u.table(table.clone());
1806 u.value(
1807 Alias::new("deleted_at"),
1808 sea_query::Value::ChronoDateTimeUtc(Some(Box::new(chrono::Utc::now()))),
1809 );
1810 for cond in &where_clauses {
1811 u.cond_where(cond.clone());
1812 }
1813 if is_sqlite {
1814 u.build_sqlx(SqliteQueryBuilder)
1815 } else {
1816 u.build_sqlx(PostgresQueryBuilder)
1817 }
1818 } else {
1819 let mut d = Query::delete();
1820 d.from_table(table.clone());
1821 for cond in &where_clauses {
1822 d.cond_where(cond.clone());
1823 }
1824 if is_sqlite {
1825 d.build_sqlx(SqliteQueryBuilder)
1826 } else {
1827 d.build_sqlx(PostgresQueryBuilder)
1828 }
1829 }
1830 };
1831
1832 let rows_affected = match tx.backend_name() {
1833 "sqlite" => {
1834 let (sql, values) = build(true);
1835 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
1836 sqlx::query_with(&sql, values)
1837 .execute(&mut **inner)
1838 .await?
1839 .rows_affected()
1840 }
1841 _ => {
1842 let (sql, values) = build(false);
1843 let inner = tx.as_pg_mut().expect("postgres backend_name");
1844 sqlx::query_with(&sql, values)
1845 .execute(&mut **inner)
1846 .await?
1847 .rows_affected()
1848 }
1849 };
1850 Ok(rows_affected)
1851 }
1852
1853 pub async fn update_json(
1857 self,
1858 body: &serde_json::Map<String, serde_json::Value>,
1859 ) -> Result<u64, crate::orm::write::WriteError> {
1860 use crate::orm::write::WriteError;
1861
1862 let needs_owned = self
1869 .meta
1870 .fields
1871 .iter()
1872 .any(|c| c.noform || c.slug_from.is_some());
1873 let mut body_owned: serde_json::Map<String, serde_json::Value>;
1874 let body: &serde_json::Map<String, serde_json::Value> = if needs_owned {
1875 body_owned = body.clone();
1876 for col in &self.meta.fields {
1877 if col.noform {
1878 body_owned.remove(&col.name);
1879 }
1880 }
1881 crate::orm::write::apply_slug_from(&self.meta.fields, &mut body_owned, true);
1882 &body_owned
1883 } else {
1884 body
1885 };
1886
1887 let validation_errors = crate::orm::validation::validate_on_update(&self.meta, body).await;
1893 if !validation_errors.is_empty() {
1894 return Err(WriteError::Multiple {
1895 errors: validation_errors,
1896 });
1897 }
1898
1899 let mut q = Query::update();
1900 q.table(crate::db::router::schema_qualified_table(&self.meta.table));
1901 let mut any = false;
1902 for col in &self.meta.fields {
1903 if col.primary_key {
1904 continue;
1905 }
1906 let Some(json) = body.get(&col.name) else {
1907 if col.auto_now {
1912 let now_value = crate::orm::write::now_for_column(col.ty);
1913 q.value(Alias::new(&col.name), now_value);
1914 any = true;
1915 }
1916 continue;
1917 };
1918 validate_numeric_bounds(col, json)?;
1919 if let (Some(fmt), Some(s)) = (col.text_format.as_deref(), json.as_str()) {
1922 if let Err(e) = crate::orm::validators::validate_text_format(fmt, s) {
1923 return Err(WriteError::Validator {
1924 field: col.name.clone(),
1925 message: e.to_string(),
1926 });
1927 }
1928 }
1929 let sea_value = crate::orm::write::json_to_sea_value(
1930 col.ty,
1931 json,
1932 col.nullable,
1933 &col.name,
1934 fk_target_pk_sql_type(col),
1935 )?;
1936 q.value(Alias::new(&col.name), sea_value);
1937 any = true;
1938 }
1939 let touches_m2m = self
1944 .meta
1945 .m2m_relations
1946 .iter()
1947 .any(|r| body.contains_key(&r.field_name));
1948 if !any && !touches_m2m {
1949 return Ok(0);
1950 }
1951 let where_clauses = self.effective_where_clauses();
1952 for cond in &where_clauses {
1953 q.cond_where(cond.clone());
1954 }
1955 let parent_pks: Vec<serde_json::Value> = match self.meta.pk_column() {
1967 Some(pk_col) => collect_parent_pks(&self.meta, pk_col, &self.where_clauses).await?,
1968 None => Vec::new(),
1969 };
1970
1971 match resolve_pool_dyn(self.meta, crate::db::RouteOp::Write) {
1972 DbPool::Sqlite(pool) => {
1973 if any {
1974 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1975 sqlx::query_with(&sql, values)
1976 .execute(&pool)
1977 .await
1978 .map_err(|e| classify_or_sqlx(e, body))?;
1979 }
1980 for pk in &parent_pks {
1981 write_m2m_junctions(&self.meta, Some(pk), body).await?;
1982 }
1983 crate::signals::emit_bulk_post_save_by_table(
1989 &self.meta.table,
1990 parent_pks.clone(),
1991 false,
1992 )
1993 .await;
1994 Ok(parent_pks.len().max(if any { 1 } else { 0 }) as u64)
1995 }
1996 DbPool::Postgres(pool) => {
1997 if any {
1998 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1999 sqlx::query_with(&sql, values)
2000 .execute(&pool)
2001 .await
2002 .map_err(|e| classify_or_sqlx(e, body))?;
2003 }
2004 for pk in &parent_pks {
2005 write_m2m_junctions(&self.meta, Some(pk), body).await?;
2006 }
2007 crate::signals::emit_bulk_post_save_by_table(
2008 &self.meta.table,
2009 parent_pks.clone(),
2010 false,
2011 )
2012 .await;
2013 Ok(parent_pks.len().max(if any { 1 } else { 0 }) as u64)
2014 }
2015 }
2016 }
2017}
2018
2019pub fn decode_to_string(
2025 row: &sqlx::sqlite::SqliteRow,
2026 col: &Column,
2027) -> Result<String, sqlx::Error> {
2028 use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
2029 use serde_json::Value;
2030 use uuid::Uuid;
2031
2032 let name = col.name.as_str();
2033 if col.nullable {
2034 return Ok(match col.ty {
2035 SqlType::SmallInt | SqlType::Integer => row
2036 .try_get::<Option<i32>, _>(name)?
2037 .map_or(String::new(), |v| v.to_string()),
2038 SqlType::BigInt => row
2039 .try_get::<Option<i64>, _>(name)?
2040 .map_or(String::new(), |v| v.to_string()),
2041 SqlType::Real => row
2042 .try_get::<Option<f32>, _>(name)?
2043 .map_or(String::new(), |v| v.to_string()),
2044 SqlType::Double => row
2045 .try_get::<Option<f64>, _>(name)?
2046 .map_or(String::new(), |v| v.to_string()),
2047 SqlType::Boolean => row
2048 .try_get::<Option<bool>, _>(name)?
2049 .map_or(String::new(), |v| {
2050 if v { "true" } else { "false" }.to_string()
2051 }),
2052 SqlType::Text => row.try_get::<Option<String>, _>(name)?.unwrap_or_default(),
2053 SqlType::Date => row
2054 .try_get::<Option<NaiveDate>, _>(name)?
2055 .map_or(String::new(), |v| v.to_string()),
2056 SqlType::Time => row
2057 .try_get::<Option<NaiveTime>, _>(name)?
2058 .map_or(String::new(), |v| v.to_string()),
2059 SqlType::Timestamptz => row
2060 .try_get::<Option<DateTime<Utc>>, _>(name)?
2061 .map_or(String::new(), |v| v.to_rfc3339()),
2062 SqlType::Uuid => row
2063 .try_get::<Option<Uuid>, _>(name)?
2064 .map_or(String::new(), |v| v.to_string()),
2065 SqlType::Json => row
2066 .try_get::<Option<Value>, _>(name)?
2067 .map_or(String::new(), |v| v.to_string()),
2068 SqlType::Array(_) => panic_array_unsupported(&col.name),
2069 SqlType::Inet
2070 | SqlType::Cidr
2071 | SqlType::MacAddr
2072 | SqlType::Xml
2073 | SqlType::Ltree
2074 | SqlType::Bit
2075 | SqlType::FullText => panic_pg_only_unsupported(&col.name),
2076 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2080 Some(SqlType::Text) => row.try_get::<Option<String>, _>(name)?.unwrap_or_default(),
2081 Some(SqlType::Uuid) => row
2082 .try_get::<Option<Uuid>, _>(name)?
2083 .map_or(String::new(), |v| v.to_string()),
2084 _ => row
2085 .try_get::<Option<i64>, _>(name)?
2086 .map_or(String::new(), |v| v.to_string()),
2087 },
2088 SqlType::Bytes => row
2089 .try_get::<Option<Vec<u8>>, _>(name)?
2090 .map_or(String::new(), |b| hex_encode(&b)),
2091 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2092 });
2093 }
2094 Ok(match col.ty {
2095 SqlType::SmallInt | SqlType::Integer => row.try_get::<i32, _>(name)?.to_string(),
2096 SqlType::BigInt => row.try_get::<i64, _>(name)?.to_string(),
2097 SqlType::Real => row.try_get::<f32, _>(name)?.to_string(),
2098 SqlType::Double => row.try_get::<f64, _>(name)?.to_string(),
2099 SqlType::Boolean => if row.try_get::<bool, _>(name)? {
2100 "true"
2101 } else {
2102 "false"
2103 }
2104 .to_string(),
2105 SqlType::Text => row.try_get::<String, _>(name)?,
2106 SqlType::Date => row.try_get::<NaiveDate, _>(name)?.to_string(),
2107 SqlType::Time => row.try_get::<NaiveTime, _>(name)?.to_string(),
2108 SqlType::Timestamptz => row.try_get::<DateTime<Utc>, _>(name)?.to_rfc3339(),
2109 SqlType::Uuid => row.try_get::<Uuid, _>(name)?.to_string(),
2110 SqlType::Json => row.try_get::<Value, _>(name)?.to_string(),
2111 SqlType::Array(_) => panic_array_unsupported(&col.name),
2112 SqlType::Inet
2113 | SqlType::Cidr
2114 | SqlType::MacAddr
2115 | SqlType::Xml
2116 | SqlType::Ltree
2117 | SqlType::Bit
2118 | SqlType::FullText => panic_pg_only_unsupported(&col.name),
2119 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2120 Some(SqlType::Text) => row.try_get::<String, _>(name)?,
2121 Some(SqlType::Uuid) => row.try_get::<Uuid, _>(name)?.to_string(),
2122 _ => row.try_get::<i64, _>(name)?.to_string(),
2123 },
2124 SqlType::Bytes => hex_encode(&row.try_get::<Vec<u8>, _>(name)?),
2125 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2126 })
2127}
2128
2129pub fn decode_pg_to_string(
2140 row: &sqlx::postgres::PgRow,
2141 col: &Column,
2142) -> Result<String, sqlx::Error> {
2143 use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
2144 use serde_json::Value;
2145 use uuid::Uuid;
2146
2147 let name = col.name.as_str();
2148 if col.nullable {
2149 return Ok(match col.ty {
2150 SqlType::SmallInt => row
2151 .try_get::<Option<i16>, _>(name)?
2152 .map_or(String::new(), |v| v.to_string()),
2153 SqlType::Integer => row
2154 .try_get::<Option<i32>, _>(name)?
2155 .map_or(String::new(), |v| v.to_string()),
2156 SqlType::BigInt => row
2157 .try_get::<Option<i64>, _>(name)?
2158 .map_or(String::new(), |v| v.to_string()),
2159 SqlType::Real => row
2160 .try_get::<Option<f32>, _>(name)?
2161 .map_or(String::new(), |v| v.to_string()),
2162 SqlType::Double => row
2163 .try_get::<Option<f64>, _>(name)?
2164 .map_or(String::new(), |v| v.to_string()),
2165 SqlType::Boolean => row
2166 .try_get::<Option<bool>, _>(name)?
2167 .map_or(String::new(), |v| {
2168 if v { "true" } else { "false" }.to_string()
2169 }),
2170 SqlType::Text => row.try_get::<Option<String>, _>(name)?.unwrap_or_default(),
2171 SqlType::Date => row
2172 .try_get::<Option<NaiveDate>, _>(name)?
2173 .map_or(String::new(), |v| v.to_string()),
2174 SqlType::Time => row
2175 .try_get::<Option<NaiveTime>, _>(name)?
2176 .map_or(String::new(), |v| v.to_string()),
2177 SqlType::Timestamptz => row
2178 .try_get::<Option<DateTime<Utc>>, _>(name)?
2179 .map_or(String::new(), |v| v.to_rfc3339()),
2180 SqlType::Uuid => row
2181 .try_get::<Option<Uuid>, _>(name)?
2182 .map_or(String::new(), |v| v.to_string()),
2183 SqlType::Json => row
2184 .try_get::<Option<Value>, _>(name)?
2185 .map_or(String::new(), |v| v.to_string()),
2186 SqlType::Array(_)
2192 | SqlType::Inet
2193 | SqlType::Cidr
2194 | SqlType::MacAddr
2195 | SqlType::Xml
2196 | SqlType::Ltree
2197 | SqlType::Bit
2198 | SqlType::FullText => row
2199 .try_get::<Option<String>, _>(name)
2200 .ok()
2201 .flatten()
2202 .unwrap_or_default(),
2203 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2206 Some(SqlType::Text) => row.try_get::<Option<String>, _>(name)?.unwrap_or_default(),
2207 Some(SqlType::Uuid) => row
2208 .try_get::<Option<Uuid>, _>(name)?
2209 .map_or(String::new(), |v| v.to_string()),
2210 _ => row
2211 .try_get::<Option<i64>, _>(name)?
2212 .map_or(String::new(), |v| v.to_string()),
2213 },
2214 SqlType::Bytes => row
2215 .try_get::<Option<Vec<u8>>, _>(name)?
2216 .map_or(String::new(), |b| hex_encode(&b)),
2217 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2218 });
2219 }
2220 Ok(match col.ty {
2221 SqlType::SmallInt => row.try_get::<i16, _>(name)?.to_string(),
2222 SqlType::Integer => row.try_get::<i32, _>(name)?.to_string(),
2223 SqlType::BigInt => row.try_get::<i64, _>(name)?.to_string(),
2224 SqlType::Real => row.try_get::<f32, _>(name)?.to_string(),
2225 SqlType::Double => row.try_get::<f64, _>(name)?.to_string(),
2226 SqlType::Boolean => if row.try_get::<bool, _>(name)? {
2227 "true"
2228 } else {
2229 "false"
2230 }
2231 .to_string(),
2232 SqlType::Text => row.try_get::<String, _>(name)?,
2233 SqlType::Date => row.try_get::<NaiveDate, _>(name)?.to_string(),
2234 SqlType::Time => row.try_get::<NaiveTime, _>(name)?.to_string(),
2235 SqlType::Timestamptz => row.try_get::<DateTime<Utc>, _>(name)?.to_rfc3339(),
2236 SqlType::Uuid => row.try_get::<Uuid, _>(name)?.to_string(),
2237 SqlType::Json => row.try_get::<Value, _>(name)?.to_string(),
2238 SqlType::Array(_)
2240 | SqlType::Inet
2241 | SqlType::Cidr
2242 | SqlType::MacAddr
2243 | SqlType::Xml
2244 | SqlType::Ltree
2245 | SqlType::Bit
2246 | SqlType::FullText => row.try_get::<String, _>(name).unwrap_or_default(),
2247 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2248 Some(SqlType::Text) => row.try_get::<String, _>(name)?,
2249 Some(SqlType::Uuid) => row.try_get::<Uuid, _>(name)?.to_string(),
2250 _ => row.try_get::<i64, _>(name)?.to_string(),
2251 },
2252 SqlType::Bytes => hex_encode(&row.try_get::<Vec<u8>, _>(name)?),
2253 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2254 })
2255}
2256
2257pub fn decode_to_json_aliased(
2270 row: &sqlx::sqlite::SqliteRow,
2271 col: &Column,
2272 alias: &str,
2273) -> Result<serde_json::Value, sqlx::Error> {
2274 let mut aliased = col.clone();
2275 aliased.name = alias.to_string();
2276 decode_to_json(row, &aliased)
2277}
2278
2279pub fn decode_pg_to_json_aliased(
2281 row: &sqlx::postgres::PgRow,
2282 col: &Column,
2283 alias: &str,
2284) -> Result<serde_json::Value, sqlx::Error> {
2285 let mut aliased = col.clone();
2286 aliased.name = alias.to_string();
2287 decode_pg_to_json(row, &aliased)
2288}
2289
2290fn fk_target_pk_sql_type(col: &Column) -> Option<SqlType> {
2314 if !matches!(col.ty, SqlType::ForeignKey) {
2315 return None;
2316 }
2317 let target_table = col.fk_target.as_deref()?;
2318 crate::migrate::pk_meta_for_table(target_table).map(|(_, ty)| ty)
2319}
2320
2321pub fn decode_to_json(
2322 row: &sqlx::sqlite::SqliteRow,
2323 col: &Column,
2324) -> Result<serde_json::Value, sqlx::Error> {
2325 use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
2326 use serde_json::Value;
2327 use uuid::Uuid;
2328
2329 let name = col.name.as_str();
2330 if col.nullable {
2331 return Ok(match col.ty {
2332 SqlType::SmallInt | SqlType::Integer => row
2333 .try_get::<Option<i32>, _>(name)?
2334 .map_or(Value::Null, Value::from),
2335 SqlType::BigInt => row
2336 .try_get::<Option<i64>, _>(name)?
2337 .map_or(Value::Null, Value::from),
2338 SqlType::Real => row
2339 .try_get::<Option<f32>, _>(name)?
2340 .map_or(Value::Null, |v| Value::from(v as f64)),
2341 SqlType::Double => row
2342 .try_get::<Option<f64>, _>(name)?
2343 .map_or(Value::Null, Value::from),
2344 SqlType::Boolean => row
2345 .try_get::<Option<bool>, _>(name)?
2346 .map_or(Value::Null, Value::from),
2347 SqlType::Text => row
2348 .try_get::<Option<String>, _>(name)?
2349 .map_or(Value::Null, Value::from),
2350 SqlType::Date => row
2351 .try_get::<Option<NaiveDate>, _>(name)?
2352 .map_or(Value::Null, |v| Value::from(v.to_string())),
2353 SqlType::Time => row
2354 .try_get::<Option<NaiveTime>, _>(name)?
2355 .map_or(Value::Null, |v| Value::from(v.to_string())),
2356 SqlType::Timestamptz => row
2357 .try_get::<Option<DateTime<Utc>>, _>(name)?
2358 .map_or(Value::Null, |v| Value::from(v.to_rfc3339())),
2359 SqlType::Uuid => row
2360 .try_get::<Option<Uuid>, _>(name)?
2361 .map_or(Value::Null, |v| Value::from(v.to_string())),
2362 SqlType::Json => row
2363 .try_get::<Option<Value>, _>(name)?
2364 .unwrap_or(Value::Null),
2365 SqlType::Array(_) => panic_array_unsupported(&col.name),
2366 SqlType::Inet
2367 | SqlType::Cidr
2368 | SqlType::MacAddr
2369 | SqlType::Xml
2370 | SqlType::Ltree
2371 | SqlType::Bit
2372 | SqlType::FullText => panic_pg_only_unsupported(&col.name),
2373 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2378 Some(SqlType::Text) => row
2379 .try_get::<Option<String>, _>(name)?
2380 .map_or(Value::Null, Value::from),
2381 Some(SqlType::Uuid) => row
2382 .try_get::<Option<Uuid>, _>(name)?
2383 .map_or(Value::Null, |v| Value::from(v.to_string())),
2384 _ => row
2385 .try_get::<Option<i64>, _>(name)?
2386 .map_or(Value::Null, Value::from),
2387 },
2388 SqlType::Bytes => row
2389 .try_get::<Option<Vec<u8>>, _>(name)?
2390 .map_or(Value::Null, |b| bytes_to_json(&b)),
2391 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2392 });
2393 }
2394 Ok(match col.ty {
2395 SqlType::SmallInt | SqlType::Integer => Value::from(row.try_get::<i32, _>(name)?),
2396 SqlType::BigInt => Value::from(row.try_get::<i64, _>(name)?),
2397 SqlType::Real => Value::from(row.try_get::<f32, _>(name)? as f64),
2398 SqlType::Double => Value::from(row.try_get::<f64, _>(name)?),
2399 SqlType::Boolean => Value::from(row.try_get::<bool, _>(name)?),
2400 SqlType::Text => Value::from(row.try_get::<String, _>(name)?),
2401 SqlType::Date => Value::from(row.try_get::<NaiveDate, _>(name)?.to_string()),
2402 SqlType::Time => Value::from(row.try_get::<NaiveTime, _>(name)?.to_string()),
2403 SqlType::Timestamptz => Value::from(row.try_get::<DateTime<Utc>, _>(name)?.to_rfc3339()),
2404 SqlType::Uuid => Value::from(row.try_get::<Uuid, _>(name)?.to_string()),
2405 SqlType::Json => row.try_get::<Value, _>(name)?,
2406 SqlType::Array(_) => panic_array_unsupported(&col.name),
2407 SqlType::Inet
2408 | SqlType::Cidr
2409 | SqlType::MacAddr
2410 | SqlType::Xml
2411 | SqlType::Ltree
2412 | SqlType::Bit
2413 | SqlType::FullText => panic_pg_only_unsupported(&col.name),
2414 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2417 Some(SqlType::Text) => Value::from(row.try_get::<String, _>(name)?),
2418 Some(SqlType::Uuid) => Value::from(row.try_get::<Uuid, _>(name)?.to_string()),
2419 _ => Value::from(row.try_get::<i64, _>(name)?),
2420 },
2421 SqlType::Bytes => bytes_to_json(&row.try_get::<Vec<u8>, _>(name)?),
2422 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2423 })
2424}
2425
2426pub fn decode_pg_to_json(
2430 row: &sqlx::postgres::PgRow,
2431 col: &Column,
2432) -> Result<serde_json::Value, sqlx::Error> {
2433 use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
2434 use serde_json::Value;
2435 use uuid::Uuid;
2436
2437 let name = col.name.as_str();
2438 if col.nullable {
2439 return Ok(match col.ty {
2440 SqlType::SmallInt => row
2441 .try_get::<Option<i16>, _>(name)?
2442 .map_or(Value::Null, Value::from),
2443 SqlType::Integer => row
2444 .try_get::<Option<i32>, _>(name)?
2445 .map_or(Value::Null, Value::from),
2446 SqlType::BigInt => row
2447 .try_get::<Option<i64>, _>(name)?
2448 .map_or(Value::Null, Value::from),
2449 SqlType::Real => row
2450 .try_get::<Option<f32>, _>(name)?
2451 .map_or(Value::Null, |v| Value::from(v as f64)),
2452 SqlType::Double => row
2453 .try_get::<Option<f64>, _>(name)?
2454 .map_or(Value::Null, Value::from),
2455 SqlType::Boolean => row
2456 .try_get::<Option<bool>, _>(name)?
2457 .map_or(Value::Null, Value::from),
2458 SqlType::Text => row
2459 .try_get::<Option<String>, _>(name)?
2460 .map_or(Value::Null, Value::from),
2461 SqlType::Date => row
2462 .try_get::<Option<NaiveDate>, _>(name)?
2463 .map_or(Value::Null, |v| Value::from(v.to_string())),
2464 SqlType::Time => row
2465 .try_get::<Option<NaiveTime>, _>(name)?
2466 .map_or(Value::Null, |v| Value::from(v.to_string())),
2467 SqlType::Timestamptz => row
2468 .try_get::<Option<DateTime<Utc>>, _>(name)?
2469 .map_or(Value::Null, |v| Value::from(v.to_rfc3339())),
2470 SqlType::Uuid => row
2471 .try_get::<Option<Uuid>, _>(name)?
2472 .map_or(Value::Null, |v| Value::from(v.to_string())),
2473 SqlType::Json => row
2474 .try_get::<Option<Value>, _>(name)?
2475 .unwrap_or(Value::Null),
2476 SqlType::Array(_)
2477 | SqlType::Inet
2478 | SqlType::Cidr
2479 | SqlType::MacAddr
2480 | SqlType::Xml
2481 | SqlType::Ltree
2482 | SqlType::Bit
2483 | SqlType::FullText => row
2484 .try_get::<Option<String>, _>(name)
2485 .ok()
2486 .flatten()
2487 .map_or(Value::Null, Value::from),
2488 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2491 Some(SqlType::Text) => row
2492 .try_get::<Option<String>, _>(name)?
2493 .map_or(Value::Null, Value::from),
2494 Some(SqlType::Uuid) => row
2495 .try_get::<Option<Uuid>, _>(name)?
2496 .map_or(Value::Null, |v| Value::from(v.to_string())),
2497 _ => row
2498 .try_get::<Option<i64>, _>(name)?
2499 .map_or(Value::Null, Value::from),
2500 },
2501 SqlType::Bytes => row
2502 .try_get::<Option<Vec<u8>>, _>(name)?
2503 .map_or(Value::Null, |b| bytes_to_json(&b)),
2504 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2505 });
2506 }
2507 Ok(match col.ty {
2508 SqlType::SmallInt => Value::from(row.try_get::<i16, _>(name)?),
2509 SqlType::Integer => Value::from(row.try_get::<i32, _>(name)?),
2510 SqlType::BigInt => Value::from(row.try_get::<i64, _>(name)?),
2511 SqlType::Real => Value::from(row.try_get::<f32, _>(name)? as f64),
2512 SqlType::Double => Value::from(row.try_get::<f64, _>(name)?),
2513 SqlType::Boolean => Value::from(row.try_get::<bool, _>(name)?),
2514 SqlType::Text => Value::from(row.try_get::<String, _>(name)?),
2515 SqlType::Date => Value::from(row.try_get::<NaiveDate, _>(name)?.to_string()),
2516 SqlType::Time => Value::from(row.try_get::<NaiveTime, _>(name)?.to_string()),
2517 SqlType::Timestamptz => Value::from(row.try_get::<DateTime<Utc>, _>(name)?.to_rfc3339()),
2518 SqlType::Uuid => Value::from(row.try_get::<Uuid, _>(name)?.to_string()),
2519 SqlType::Json => row.try_get::<Value, _>(name)?,
2520 SqlType::Array(_)
2521 | SqlType::Inet
2522 | SqlType::Cidr
2523 | SqlType::MacAddr
2524 | SqlType::Xml
2525 | SqlType::Ltree
2526 | SqlType::Bit
2527 | SqlType::FullText => row
2528 .try_get::<String, _>(name)
2529 .map(Value::from)
2530 .unwrap_or(Value::Null),
2531 SqlType::ForeignKey => match fk_target_pk_sql_type(col) {
2534 Some(SqlType::Text) => Value::from(row.try_get::<String, _>(name)?),
2535 Some(SqlType::Uuid) => Value::from(row.try_get::<Uuid, _>(name)?.to_string()),
2536 _ => Value::from(row.try_get::<i64, _>(name)?),
2537 },
2538 SqlType::Bytes => bytes_to_json(&row.try_get::<Vec<u8>, _>(name)?),
2539 SqlType::Decimal => panic_pg_only_unsupported(&col.name),
2540 })
2541}
2542
2543fn form_str_to_sea_value(col: &Column, raw: &str) -> Result<SeaValue, WriteError> {
2551 if raw.is_empty() {
2552 if col.ty == SqlType::Boolean {
2553 return Ok(SeaValue::Bool(Some(false)));
2555 }
2556 if col.nullable {
2557 return Ok(null_for(col.ty));
2558 }
2559 return Err(WriteError::RequiredFieldMissing {
2560 field: col.name.clone(),
2561 });
2562 }
2563 if matches!(col.ty, SqlType::Json | SqlType::Array(_)) {
2575 let parsed: serde_json::Value =
2576 serde_json::from_str(raw).map_err(|e| WriteError::Validator {
2577 field: col.name.clone(),
2578 message: format!("Not valid JSON: {e}"),
2579 })?;
2580 return json_to_sea_value(col.ty, &parsed, col.nullable, &col.name, None);
2581 }
2582 if matches!(col.ty, SqlType::ForeignKey) {
2583 return match fk_target_pk_sql_type(col) {
2584 Some(SqlType::Text) => Ok(SeaValue::String(Some(Box::new(raw.to_string())))),
2585 Some(SqlType::Uuid) => uuid::Uuid::parse_str(raw)
2586 .map(|v| SeaValue::Uuid(Some(Box::new(v))))
2587 .map_err(|_| WriteError::TypeMismatch {
2588 field: col.name.clone(),
2589 expected: SqlType::Uuid,
2590 got: raw.to_string(),
2591 }),
2592 _ => raw
2593 .parse::<i64>()
2594 .map(|v| SeaValue::BigInt(Some(v)))
2595 .map_err(|_| WriteError::TypeMismatch {
2596 field: col.name.clone(),
2597 expected: SqlType::BigInt,
2598 got: raw.to_string(),
2599 }),
2600 };
2601 }
2602 let json = serde_json::Value::String(raw.to_string());
2603 json_to_sea_value(col.ty, &json, col.nullable, &col.name, None)
2604}
2605
2606fn hex_encode(bytes: &[u8]) -> String {
2610 let mut out = String::with_capacity(bytes.len() * 2);
2611 for b in bytes {
2612 out.push_str(&format!("{b:02x}"));
2613 }
2614 out
2615}
2616
2617fn bytes_to_json(bytes: &[u8]) -> serde_json::Value {
2620 serde_json::Value::Array(bytes.iter().map(|b| serde_json::Value::from(*b)).collect())
2621}
2622
2623fn panic_array_unsupported(column: &str) -> ! {
2624 panic!(
2625 "DynQuerySet: column `{column}` is a Postgres-only Array; the \
2626 field/backend system check should have failed boot."
2627 )
2628}
2629
2630fn panic_pg_only_unsupported(column: &str) -> ! {
2631 panic!(
2632 "DynQuerySet: column `{column}` is a Postgres-only network type \
2633 (Inet/Cidr/MacAddr); the field/backend system check should \
2634 have failed boot."
2635 )
2636}
2637
2638fn classify_or_sqlx(
2644 e: sqlx::Error,
2645 body: &serde_json::Map<String, serde_json::Value>,
2646) -> crate::orm::write::WriteError {
2647 if let Some(classified) = crate::orm::validation::classify_sql_error(&e, body) {
2648 return classified;
2649 }
2650 crate::orm::write::WriteError::Sqlx(e)
2651}
2652
2653fn validate_numeric_bounds(
2654 col: &Column,
2655 json: &serde_json::Value,
2656) -> Result<(), crate::orm::write::WriteError> {
2657 let Some(n) = json.as_f64() else {
2658 return Ok(());
2659 };
2660 if let Some(min) = col.min {
2661 if n < min as f64 {
2662 return Err(crate::orm::write::WriteError::Validator {
2663 field: col.name.clone(),
2664 message: format!("must be >= {min} (got {n})."),
2665 });
2666 }
2667 }
2668 if let Some(max) = col.max {
2669 if n > max as f64 {
2670 return Err(crate::orm::write::WriteError::Validator {
2671 field: col.name.clone(),
2672 message: format!("must be <= {max} (got {n})."),
2673 });
2674 }
2675 }
2676 Ok(())
2677}
2678
2679fn json_pk_to_sea(v: &serde_json::Value) -> Option<sea_query::Value> {
2685 match v {
2686 serde_json::Value::Number(n) => n.as_i64().map(|i| sea_query::Value::BigInt(Some(i))),
2687 serde_json::Value::String(s) => Some(sea_query::Value::String(Some(Box::new(s.clone())))),
2688 _ => None,
2689 }
2690}
2691
2692fn normalize_sr_token(name: &str) -> String {
2709 name.replace("__", ".")
2710}
2711
2712fn validate_sr_chain(root_meta: &crate::migrate::ModelMeta, chain: &str) -> Option<Vec<String>> {
2722 let hops: Vec<&str> = chain.split('.').filter(|s| !s.is_empty()).collect();
2723 if hops.is_empty() {
2724 return None;
2725 }
2726 let registered = crate::migrate::registered_models();
2727 let mut targets: Vec<String> = Vec::with_capacity(hops.len());
2728 let mut current_table: String = root_meta.table.clone();
2729 let mut current_meta: Option<crate::migrate::ModelMeta> = None;
2730 for hop in &hops {
2731 let meta_ref: &crate::migrate::ModelMeta =
2732 if current_table == root_meta.table && current_meta.is_none() {
2733 root_meta
2734 } else {
2735 current_meta = registered
2736 .iter()
2737 .find(|m| m.table == current_table)
2738 .cloned();
2739 current_meta.as_ref()?
2740 };
2741 let col = meta_ref.fields.iter().find(|c| &c.name == hop)?;
2742 let target = col.fk_target.clone()?;
2743 targets.push(target.clone());
2744 current_table = target;
2745 }
2746 Some(targets)
2747}
2748
2749async fn hydrate_select_related_into(
2766 meta: &crate::migrate::ModelMeta,
2767 sr_fields: &[String],
2768 rows: &mut [serde_json::Map<String, serde_json::Value>],
2769) -> Result<(), sqlx::Error> {
2770 let pool = resolve_pool_dyn(meta, crate::db::RouteOp::Read);
2771 for chain in sr_fields {
2772 let hops: Vec<&str> = chain.split('.').filter(|s| !s.is_empty()).collect();
2773 if hops.is_empty() {
2774 continue;
2775 }
2776 let Some(targets) = validate_sr_chain(meta, chain) else {
2777 continue;
2782 };
2783
2784 let registered = crate::migrate::registered_models();
2793 let hop_target_pk: Vec<(String, SqlType)> = targets
2794 .iter()
2795 .filter_map(|t| {
2796 registered
2797 .iter()
2798 .find(|m| &m.table == t)
2799 .and_then(|m| m.pk_column().map(|c| (c.name.clone(), c.ty)))
2800 })
2801 .collect();
2802 if hop_target_pk.len() != hops.len() {
2803 continue;
2807 }
2808 let hop_target_soft_delete: Vec<bool> = targets
2809 .iter()
2810 .map(|t| {
2811 registered
2812 .iter()
2813 .find(|m| &m.table == t)
2814 .is_some_and(|m| m.soft_delete)
2815 })
2816 .collect();
2817
2818 let first_field = hops[0];
2822 let mut ids: Vec<serde_json::Value> = Vec::with_capacity(rows.len());
2823 for row in rows.iter() {
2824 let Some(v) = row.get(first_field) else {
2825 continue;
2826 };
2827 if v.is_null() {
2828 continue;
2829 }
2830 ids.push(v.clone());
2831 }
2832 if ids.is_empty() {
2833 continue;
2834 }
2835 dedup_by_pk_key(&mut ids);
2836 let mut levels: Vec<Vec<serde_json::Value>> = Vec::with_capacity(hops.len());
2837 levels.push(
2838 crate::orm::queryset::hydration::fetch_related_as_json_by_pk(
2839 &targets[0],
2840 &hop_target_pk[0].0,
2841 hop_target_pk[0].1,
2842 hop_target_soft_delete[0],
2843 &ids,
2844 &pool,
2845 )
2846 .await?,
2847 );
2848
2849 for hop_idx in 1..hops.len() {
2850 let hop_field = hops[hop_idx];
2851 let hop_target = &targets[hop_idx];
2852 let prev_lvl = &levels[hop_idx - 1];
2853 let mut next_ids: Vec<serde_json::Value> = prev_lvl
2854 .iter()
2855 .filter_map(|r| {
2856 let v = r.as_object()?.get(hop_field)?;
2857 if v.is_null() { None } else { Some(v.clone()) }
2858 })
2859 .collect();
2860 if next_ids.is_empty() {
2861 break;
2866 }
2867 dedup_by_pk_key(&mut next_ids);
2868 levels.push(
2869 crate::orm::queryset::hydration::fetch_related_as_json_by_pk(
2870 hop_target,
2871 &hop_target_pk[hop_idx].0,
2872 hop_target_pk[hop_idx].1,
2873 hop_target_soft_delete[hop_idx],
2874 &next_ids,
2875 &pool,
2876 )
2877 .await?,
2878 );
2879 }
2880
2881 if levels.len() > 1 {
2886 for i in (0..levels.len() - 1).rev() {
2887 let next_pk_col = &hop_target_pk[i + 1].0;
2888 let next_by_pk: HashMap<String, serde_json::Value> = levels[i + 1]
2889 .iter()
2890 .filter_map(|obj| {
2891 let map = obj.as_object()?;
2892 let pk_val = map.get(next_pk_col.as_str())?;
2893 Some((pk_json_key(pk_val), obj.clone()))
2894 })
2895 .collect();
2896 let hop_field = hops[i + 1];
2897 for row in levels[i].iter_mut() {
2898 let Some(map) = row.as_object_mut() else {
2899 continue;
2900 };
2901 let Some(fk_val) = map.get(hop_field) else {
2902 continue;
2903 };
2904 if fk_val.is_null() {
2905 continue;
2906 }
2907 let key = pk_json_key(fk_val);
2908 if let Some(next_json) = next_by_pk.get(&key) {
2909 map.insert(hop_field.to_string(), next_json.clone());
2910 }
2911 }
2912 }
2913 }
2914
2915 let first_pk_col = &hop_target_pk[0].0;
2922 let first_by_pk: HashMap<String, serde_json::Value> = levels
2923 .into_iter()
2924 .next()
2925 .unwrap_or_default()
2926 .into_iter()
2927 .filter_map(|obj| {
2928 let map = obj.as_object()?;
2929 let pk_val = map.get(first_pk_col.as_str())?;
2930 Some((pk_json_key(pk_val), obj.clone()))
2931 })
2932 .collect();
2933 for row in rows.iter_mut() {
2934 let Some(fk_val) = row.get(first_field) else {
2935 continue;
2936 };
2937 if fk_val.is_null() {
2938 continue;
2939 }
2940 let key = pk_json_key(fk_val);
2941 if let Some(resolved) = first_by_pk.get(&key) {
2942 row.insert(first_field.to_string(), resolved.clone());
2943 }
2944 }
2945 }
2946 Ok(())
2947}
2948
2949fn dedup_by_pk_key(ids: &mut Vec<serde_json::Value>) {
2954 let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
2955 ids.retain(|v| seen.insert(pk_json_key(v)));
2956}
2957
2958async fn hydrate_m2m_batched(
2972 meta: &crate::migrate::ModelMeta,
2973 pk_name: &str,
2974 rows: &mut [serde_json::Map<String, serde_json::Value>],
2975) -> Result<(), sqlx::Error> {
2976 if meta.m2m_relations.is_empty() || rows.is_empty() {
2977 return Ok(());
2978 }
2979
2980 for row in rows.iter_mut() {
2985 for rel in &meta.m2m_relations {
2986 row.insert(rel.field_name.clone(), serde_json::Value::Array(Vec::new()));
2987 }
2988 }
2989
2990 let mut parent_sea_vals: Vec<sea_query::Value> = Vec::with_capacity(rows.len());
2994 let mut seen_keys: std::collections::HashSet<String> = std::collections::HashSet::new();
2995 for row in rows.iter() {
2996 let Some(pk_json) = row.get(pk_name) else {
2997 continue;
2998 };
2999 let Some(sea_val) = json_pk_to_sea(pk_json) else {
3000 continue;
3001 };
3002 let key = pk_json_key(pk_json);
3003 if seen_keys.insert(key) {
3004 parent_sea_vals.push(sea_val);
3005 }
3006 }
3007 if parent_sea_vals.is_empty() {
3008 return Ok(());
3009 }
3010
3011 for rel in &meta.m2m_relations {
3012 let junction_table = format!("{}_{}", meta.table, rel.field_name);
3013 let mut sel = Query::select();
3014 sel.from(crate::db::router::schema_qualified_table(&junction_table));
3015 sel.column(Alias::new("parent_id"));
3016 sel.column(Alias::new("child_id"));
3017 sel.and_where(Expr::col(Alias::new("parent_id")).is_in(parent_sea_vals.clone()));
3018
3019 let mut children_by_parent: HashMap<String, Vec<serde_json::Value>> = HashMap::new();
3020 match resolve_pool_dyn(meta, crate::db::RouteOp::Read) {
3021 DbPool::Sqlite(pool) => {
3022 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
3023 let db_rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
3024 for r in &db_rows {
3025 let parent = read_junction_id_sqlite(r, "parent_id")?;
3026 let child = read_junction_id_sqlite(r, "child_id")?;
3027 children_by_parent
3028 .entry(pk_json_key(&parent))
3029 .or_default()
3030 .push(child);
3031 }
3032 }
3033 DbPool::Postgres(pool) => {
3034 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
3035 let db_rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
3036 for r in &db_rows {
3037 let parent = read_junction_id_pg(r, "parent_id")?;
3038 let child = read_junction_id_pg(r, "child_id")?;
3039 children_by_parent
3040 .entry(pk_json_key(&parent))
3041 .or_default()
3042 .push(child);
3043 }
3044 }
3045 }
3046
3047 for row in rows.iter_mut() {
3048 let Some(pk_json) = row.get(pk_name) else {
3049 continue;
3050 };
3051 let key = pk_json_key(pk_json);
3052 if let Some(children) = children_by_parent.remove(&key) {
3053 row.insert(rel.field_name.clone(), serde_json::Value::Array(children));
3054 }
3055 }
3056 }
3057 Ok(())
3058}
3059
3060fn pk_json_key(v: &serde_json::Value) -> String {
3066 match v {
3067 serde_json::Value::Number(n) => format!("n:{n}"),
3068 serde_json::Value::String(s) => format!("s:{s}"),
3069 other => format!("o:{other}"),
3070 }
3071}
3072
3073fn read_junction_id_sqlite(
3078 row: &sqlx::sqlite::SqliteRow,
3079 col: &str,
3080) -> Result<serde_json::Value, sqlx::Error> {
3081 if let Ok(i) = row.try_get::<i64, _>(col) {
3082 return Ok(serde_json::Value::Number(i.into()));
3083 }
3084 let s = row.try_get::<String, _>(col)?;
3085 Ok(serde_json::Value::String(s))
3086}
3087
3088fn read_junction_id_pg(
3089 row: &sqlx::postgres::PgRow,
3090 col: &str,
3091) -> Result<serde_json::Value, sqlx::Error> {
3092 if let Ok(i) = row.try_get::<i64, _>(col) {
3093 return Ok(serde_json::Value::Number(i.into()));
3094 }
3095 let s = row.try_get::<String, _>(col)?;
3096 Ok(serde_json::Value::String(s))
3097}
3098
3099async fn hydrate_m2m_into(
3100 meta: &crate::migrate::ModelMeta,
3101 parent_pk_json: Option<&serde_json::Value>,
3102 out: &mut serde_json::Map<String, serde_json::Value>,
3103) -> Result<(), sqlx::Error> {
3104 if meta.m2m_relations.is_empty() {
3105 return Ok(());
3106 }
3107 let Some(parent_pk_value) = parent_pk_json.and_then(json_pk_to_sea) else {
3108 return Ok(());
3109 };
3110 for rel in &meta.m2m_relations {
3111 let junction_table = format!("{}_{}", meta.table, rel.field_name);
3112 let mut sel = Query::select();
3113 sel.from(crate::db::router::schema_qualified_table(&junction_table));
3114 sel.column(Alias::new("child_id"));
3115 sel.and_where(Expr::col(Alias::new("parent_id")).eq(parent_pk_value.clone()));
3116 let children: Vec<serde_json::Value> =
3117 match resolve_pool_dyn(meta, crate::db::RouteOp::Read) {
3118 DbPool::Sqlite(pool) => {
3119 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
3120 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
3121 rows.iter()
3122 .map(|r| {
3123 r.try_get::<i64, _>("child_id")
3124 .map(|i| serde_json::Value::Number(i.into()))
3125 .or_else(|_| {
3126 r.try_get::<String, _>("child_id")
3127 .map(serde_json::Value::String)
3128 })
3129 })
3130 .collect::<Result<Vec<_>, _>>()?
3131 }
3132 DbPool::Postgres(pool) => {
3133 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
3134 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
3135 rows.iter()
3136 .map(|r| {
3137 r.try_get::<i64, _>("child_id")
3138 .map(|i| serde_json::Value::Number(i.into()))
3139 .or_else(|_| {
3140 r.try_get::<String, _>("child_id")
3141 .map(serde_json::Value::String)
3142 })
3143 })
3144 .collect::<Result<Vec<_>, _>>()?
3145 }
3146 };
3147 out.insert(rel.field_name.clone(), serde_json::Value::Array(children));
3148 }
3149 Ok(())
3150}
3151
3152async fn collect_parent_pks(
3159 meta: &crate::migrate::ModelMeta,
3160 pk_col: &crate::migrate::Column,
3161 where_clauses: &[Condition],
3162) -> Result<Vec<serde_json::Value>, crate::orm::write::WriteError> {
3163 let mut sel = Query::select();
3164 sel.from(crate::db::router::schema_qualified_table(&meta.table));
3165 sel.column(Alias::new(&pk_col.name));
3166 for cond in where_clauses {
3167 sel.cond_where(cond.clone());
3168 }
3169 match resolve_pool_dyn(meta, crate::db::RouteOp::Read) {
3170 DbPool::Sqlite(pool) => {
3171 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
3172 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
3173 rows.iter()
3174 .map(|row| decode_to_json(row, pk_col))
3175 .collect::<Result<Vec<_>, _>>()
3176 .map_err(crate::orm::write::WriteError::Sqlx)
3177 }
3178 DbPool::Postgres(pool) => {
3179 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
3180 let rows = sqlx::query_with(&sql, values).fetch_all(&pool).await?;
3181 rows.iter()
3182 .map(|row| decode_pg_to_json(row, pk_col))
3183 .collect::<Result<Vec<_>, _>>()
3184 .map_err(crate::orm::write::WriteError::Sqlx)
3185 }
3186 }
3187}
3188
3189async fn collect_parent_pks_in_tx(
3193 meta: &crate::migrate::ModelMeta,
3194 pk_col: &crate::migrate::Column,
3195 where_clauses: &[Condition],
3196 tx: &mut crate::db::Transaction,
3197) -> Result<Vec<serde_json::Value>, crate::orm::write::WriteError> {
3198 let mut sel = Query::select();
3199 sel.from(crate::db::router::schema_qualified_table(&meta.table));
3200 sel.column(Alias::new(&pk_col.name));
3201 for cond in where_clauses {
3202 sel.cond_where(cond.clone());
3203 }
3204 match tx.backend_name() {
3205 "sqlite" => {
3206 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
3207 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
3208 let rows = sqlx::query_with(&sql, values)
3209 .fetch_all(&mut **inner)
3210 .await?;
3211 rows.iter()
3212 .map(|row| decode_to_json(row, pk_col))
3213 .collect::<Result<Vec<_>, _>>()
3214 .map_err(crate::orm::write::WriteError::Sqlx)
3215 }
3216 _ => {
3217 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
3218 let inner = tx.as_pg_mut().expect("postgres backend_name");
3219 let rows = sqlx::query_with(&sql, values)
3220 .fetch_all(&mut **inner)
3221 .await?;
3222 rows.iter()
3223 .map(|row| decode_pg_to_json(row, pk_col))
3224 .collect::<Result<Vec<_>, _>>()
3225 .map_err(crate::orm::write::WriteError::Sqlx)
3226 }
3227 }
3228}
3229
3230fn normalise_insert_body(
3247 meta: &crate::migrate::ModelMeta,
3248 body: &serde_json::Map<String, serde_json::Value>,
3249) -> Option<serde_json::Map<String, serde_json::Value>> {
3250 let needs_owned = meta
3251 .fields
3252 .iter()
3253 .any(|c| c.noform || c.slug_from.is_some());
3254 if !needs_owned {
3255 return None;
3256 }
3257 let mut owned = body.clone();
3258 for col in &meta.fields {
3259 if col.noform {
3260 owned.remove(&col.name);
3261 }
3262 }
3263 crate::orm::write::apply_slug_from(&meta.fields, &mut owned, false);
3264 Some(owned)
3265}
3266
3267struct InsertPlan {
3269 q: sea_query::InsertStatement,
3270 pk_name: String,
3271 pk_ty: SqlType,
3272}
3273
3274fn build_insert_plan(
3283 meta: &crate::migrate::ModelMeta,
3284 body: &serde_json::Map<String, serde_json::Value>,
3285) -> Result<InsertPlan, crate::orm::write::WriteError> {
3286 use crate::orm::write::{WriteError, is_default_pk};
3287
3288 let mut cols: Vec<&str> = Vec::new();
3289 let mut values: Vec<SeaValue> = Vec::new();
3290 for col in &meta.fields {
3291 if col.primary_key {
3292 let supplied = body.get(&col.name);
3293 let is_sentinel = match supplied {
3294 None | Some(serde_json::Value::Null) => true,
3295 Some(v) => is_default_pk(col.ty, v),
3296 };
3297 if matches!(
3298 col.ty,
3299 SqlType::Integer | SqlType::BigInt | SqlType::SmallInt
3300 ) && is_sentinel
3301 {
3302 continue;
3303 }
3304 }
3305 let Some(json) = body.get(&col.name) else {
3306 if col.auto_now_add || col.auto_now {
3307 let now_value = crate::orm::write::now_for_column(col.ty);
3308 cols.push(&col.name);
3309 values.push(now_value);
3310 continue;
3311 }
3312 continue;
3313 };
3314 if json.is_null() {
3315 continue;
3316 }
3317 validate_numeric_bounds(col, json)?;
3318 if let (Some(fmt), Some(s)) = (col.text_format.as_deref(), json.as_str()) {
3319 if let Err(e) = crate::orm::validators::validate_text_format(fmt, s) {
3320 return Err(WriteError::Validator {
3321 field: col.name.clone(),
3322 message: e.to_string(),
3323 });
3324 }
3325 }
3326 let sea_value = crate::orm::write::json_to_sea_value(
3327 col.ty,
3328 json,
3329 col.nullable,
3330 &col.name,
3331 fk_target_pk_sql_type(col),
3332 )?;
3333 cols.push(&col.name);
3334 values.push(sea_value);
3335 }
3336
3337 let pk_col = meta.fields.iter().find(|c| c.primary_key).ok_or_else(|| {
3338 WriteError::Sqlx(sqlx::Error::Protocol(
3339 "insert_json: model has no PK".to_string(),
3340 ))
3341 })?;
3342 let pk_name = pk_col.name.clone();
3343 let pk_ty = pk_col.ty;
3344
3345 let mut q = Query::insert();
3346 q.into_table(crate::db::router::schema_qualified_table(&meta.table));
3347 q.columns(cols.iter().map(|c| Alias::new(*c)).collect::<Vec<_>>());
3348 let exprs: Vec<sea_query::SimpleExpr> = values.into_iter().map(Into::into).collect();
3349 q.values_panic(exprs);
3350
3351 Ok(InsertPlan { q, pk_name, pk_ty })
3352}
3353
3354async fn write_m2m_junctions_in_tx(
3358 meta: &crate::migrate::ModelMeta,
3359 parent_pk_json: Option<&serde_json::Value>,
3360 body: &serde_json::Map<String, serde_json::Value>,
3361 tx: &mut crate::db::Transaction,
3362) -> Result<(), crate::orm::write::WriteError> {
3363 if meta.m2m_relations.is_empty() {
3364 return Ok(());
3365 }
3366 let Some(parent_pk_value) = parent_pk_json.and_then(json_pk_to_sea) else {
3367 return Ok(());
3368 };
3369 for rel in &meta.m2m_relations {
3370 let Some(value) = body.get(&rel.field_name) else {
3371 continue;
3372 };
3373 let Some(items) = value.as_array() else {
3374 continue;
3375 };
3376 let mut child_ids: Vec<sea_query::Value> = Vec::with_capacity(items.len());
3377 for item in items {
3378 if item.is_null() {
3379 continue;
3380 }
3381 if let Some(v) = json_pk_to_sea(item) {
3382 child_ids.push(v);
3383 }
3384 }
3385 let junction_table = format!("{}_{}", meta.table, rel.field_name);
3386 crate::orm::m2m::set_junction_dynamic_in_tx(
3387 &junction_table,
3388 parent_pk_value.clone(),
3389 child_ids,
3390 tx,
3391 )
3392 .await
3393 .map_err(crate::orm::write::WriteError::Sqlx)?;
3394 }
3395 Ok(())
3396}
3397
3398async fn hydrate_m2m_into_tx(
3403 meta: &crate::migrate::ModelMeta,
3404 parent_pk_json: Option<&serde_json::Value>,
3405 out: &mut serde_json::Map<String, serde_json::Value>,
3406 tx: &mut crate::db::Transaction,
3407) -> Result<(), sqlx::Error> {
3408 if meta.m2m_relations.is_empty() {
3409 return Ok(());
3410 }
3411 let Some(parent_pk_value) = parent_pk_json.and_then(json_pk_to_sea) else {
3412 return Ok(());
3413 };
3414 for rel in &meta.m2m_relations {
3415 let junction_table = format!("{}_{}", meta.table, rel.field_name);
3416 let mut sel = Query::select();
3417 sel.from(crate::db::router::schema_qualified_table(&junction_table));
3418 sel.column(Alias::new("child_id"));
3419 sel.and_where(Expr::col(Alias::new("parent_id")).eq(parent_pk_value.clone()));
3420 let children: Vec<serde_json::Value> = match tx.backend_name() {
3421 "sqlite" => {
3422 let inner = tx.as_sqlite_mut().expect("sqlite backend_name");
3423 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
3424 let rows = sqlx::query_with(&sql, values)
3425 .fetch_all(&mut **inner)
3426 .await?;
3427 rows.iter()
3428 .map(|r| {
3429 r.try_get::<i64, _>("child_id")
3430 .map(|i| serde_json::Value::Number(i.into()))
3431 .or_else(|_| {
3432 r.try_get::<String, _>("child_id")
3433 .map(serde_json::Value::String)
3434 })
3435 })
3436 .collect::<Result<Vec<_>, _>>()?
3437 }
3438 _ => {
3439 let inner = tx.as_pg_mut().expect("postgres backend_name");
3440 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
3441 let rows = sqlx::query_with(&sql, values)
3442 .fetch_all(&mut **inner)
3443 .await?;
3444 rows.iter()
3445 .map(|r| {
3446 r.try_get::<i64, _>("child_id")
3447 .map(|i| serde_json::Value::Number(i.into()))
3448 .or_else(|_| {
3449 r.try_get::<String, _>("child_id")
3450 .map(serde_json::Value::String)
3451 })
3452 })
3453 .collect::<Result<Vec<_>, _>>()?
3454 }
3455 };
3456 out.insert(rel.field_name.clone(), serde_json::Value::Array(children));
3457 }
3458 Ok(())
3459}
3460
3461async fn write_m2m_junctions(
3462 meta: &crate::migrate::ModelMeta,
3463 parent_pk_json: Option<&serde_json::Value>,
3464 body: &serde_json::Map<String, serde_json::Value>,
3465) -> Result<(), crate::orm::write::WriteError> {
3466 if meta.m2m_relations.is_empty() {
3467 return Ok(());
3468 }
3469 let Some(parent_pk_value) = parent_pk_json.and_then(json_pk_to_sea) else {
3470 return Ok(());
3471 };
3472 for rel in &meta.m2m_relations {
3473 let Some(value) = body.get(&rel.field_name) else {
3474 continue;
3475 };
3476 let Some(items) = value.as_array() else {
3477 continue; };
3479 let mut child_ids: Vec<sea_query::Value> = Vec::with_capacity(items.len());
3480 for item in items {
3481 if item.is_null() {
3482 continue;
3483 }
3484 if let Some(v) = json_pk_to_sea(item) {
3485 child_ids.push(v);
3486 }
3487 }
3488 let junction_table = format!("{}_{}", meta.table, rel.field_name);
3489 crate::orm::m2m::set_junction_dynamic(
3490 &junction_table,
3491 parent_pk_value.clone(),
3492 child_ids,
3493 Some(&meta.name),
3494 )
3495 .await
3496 .map_err(crate::orm::write::WriteError::Sqlx)?;
3497 }
3498 Ok(())
3499}
3500
3501fn coerce_csv_cell(ty: SqlType, nullable: bool, raw: &str) -> serde_json::Value {
3518 use serde_json::Value;
3519 if raw.is_empty() && nullable {
3520 return Value::Null;
3521 }
3522 match ty {
3523 SqlType::SmallInt | SqlType::Integer | SqlType::BigInt | SqlType::ForeignKey => raw
3524 .parse::<i64>()
3525 .map(Value::from)
3526 .unwrap_or_else(|_| Value::String(raw.to_string())),
3527 SqlType::Real | SqlType::Double => raw
3528 .parse::<f64>()
3529 .ok()
3530 .and_then(serde_json::Number::from_f64)
3531 .map(Value::Number)
3532 .unwrap_or_else(|| Value::String(raw.to_string())),
3533 SqlType::Boolean => match raw.trim().to_ascii_lowercase().as_str() {
3534 "true" | "1" | "t" | "yes" | "y" => Value::Bool(true),
3535 "false" | "0" | "f" | "no" | "n" => Value::Bool(false),
3536 _ => Value::String(raw.to_string()),
3537 },
3538 SqlType::Json => {
3539 serde_json::from_str(raw).unwrap_or_else(|_| Value::String(raw.to_string()))
3540 }
3541 _ => Value::String(raw.to_string()),
3542 }
3543}
3544
3545#[derive(Debug, Default)]
3552pub struct CsvImportReport {
3553 pub inserted: usize,
3554 pub errors: Vec<(usize, String)>,
3555}
3556
3557pub async fn import_table_rows(
3568 meta: &ModelMeta,
3569 headers: &[String],
3570 rows: &[Vec<String>],
3571) -> CsvImportReport {
3572 let col_for: HashMap<&str, &Column> =
3573 meta.fields.iter().map(|c| (c.name.as_str(), c)).collect();
3574
3575 let mut report = CsvImportReport::default();
3576 for (i, row) in rows.iter().enumerate() {
3577 let mut obj = serde_json::Map::new();
3578 for (header, cell) in headers.iter().zip(row.iter()) {
3579 if let Some(col) = col_for.get(header.as_str()) {
3580 obj.insert(header.clone(), coerce_csv_cell(col.ty, col.nullable, cell));
3581 }
3582 }
3583 match DynQuerySet::for_meta(meta).insert_json(&obj).await {
3584 Ok(_) => report.inserted += 1,
3585 Err(e) => report.errors.push((i + 2, e.to_string())),
3586 }
3587 }
3588 report
3589}
3590
3591#[cfg(test)]
3592mod tests {
3593 use super::form_str_to_sea_value;
3594 use crate::migrate::Column;
3595 use crate::orm::{FkAction, SqlType};
3596 use sea_query::Value as SeaValue;
3597
3598 fn col(name: &str, ty: SqlType, nullable: bool) -> Column {
3599 Column {
3600 name: name.to_string(),
3601 ty,
3602 primary_key: false,
3603 nullable,
3604 fk_target: None,
3605 noform: false,
3606 db_constraint: true,
3607 noedit: false,
3608 is_string_repr: false,
3609 max_length: 0,
3610 choices: Vec::new(),
3611 choice_labels: Vec::new(),
3612 default: String::new(),
3613 is_multichoice: false,
3614 unique: false,
3615 on_delete: FkAction::NoAction,
3616 on_update: FkAction::NoAction,
3617 index: false,
3618 auto_now_add: false,
3619 auto_now: false,
3620 help: String::new(),
3621 example: String::new(),
3622 widget: None,
3623 supported_backends: Vec::new(),
3624 min: None,
3625 max: None,
3626 text_format: None,
3627 slug_from: None,
3628 }
3629 }
3630
3631 #[test]
3632 fn form_fk_numeric_string_binds_as_bigint() {
3633 let mut plugin = col("plugin", SqlType::ForeignKey, false);
3634 plugin.fk_target = Some("plugin".to_string());
3635
3636 let value = form_str_to_sea_value(&plugin, "1").expect("coerce FK id");
3637
3638 assert_eq!(
3639 value,
3640 SeaValue::BigInt(Some(1)),
3641 "integer-backed FK form values must bind as bigint, not text"
3642 );
3643 }
3644
3645 #[test]
3646 fn nullable_form_fk_blank_binds_as_null_bigint() {
3647 let mut parent = col("parent", SqlType::ForeignKey, true);
3648 parent.fk_target = Some("plugin_comment".to_string());
3649
3650 let value = form_str_to_sea_value(&parent, "").expect("blank nullable FK");
3651
3652 assert_eq!(
3653 value,
3654 SeaValue::BigInt(None),
3655 "blank nullable integer-backed FK should bind SQL NULL"
3656 );
3657 }
3658}