fletch_orm/
ops.rs

1//! Dialect-generic CRUD operations.
2//!
3//! Core functions accept any [`Executor`] so the same CRUD runs against a pool
4//! or an in-flight transaction. [`Pool`] and [`Transaction`] delegate to these
5//! functions.
6
7use sqlx::{Database, Executor, FromRow};
8
9use crate::bind::DatabaseBindable;
10use crate::built_query::BuiltQuery;
11use crate::column::Column;
12use crate::dialect::Dialect;
13use crate::entity::Entity;
14use crate::error::FletchError;
15use crate::filter::{Filter, Op};
16#[cfg(any(feature = "sqlite", feature = "postgres", feature = "mysql"))]
17use crate::pool::{Pool, Transaction};
18use crate::query_builder::QueryBuilder;
19use crate::value::Value;
20
21/// Find an entity by its primary key.
22pub async fn find_by_id<'e, E, DB, D, Ex>(
23    dialect: &D,
24    executor: Ex,
25    id: E::Id,
26) -> Result<E, FletchError>
27where
28    E: Entity,
29    D: Dialect,
30    DB: Database + DatabaseBindable,
31    Ex: Executor<'e, Database = DB> + Send,
32    E: for<'r> FromRow<'r, <DB as Database>::Row> + Send + Unpin,
33{
34    let query = QueryBuilder::select(dialect, E::table_name())
35        .columns(&column_names::<E>())
36        .filter(Filter::new(E::id_column(), Op::Eq, id))
37        .build();
38
39    query.fetch_one::<DB, E, _>(executor).await
40}
41
42/// Fetch all entities of this type.
43pub async fn find_all<'e, E, DB, D, Ex>(dialect: &D, executor: Ex) -> Result<Vec<E>, FletchError>
44where
45    E: Entity,
46    D: Dialect,
47    DB: Database + DatabaseBindable,
48    Ex: Executor<'e, Database = DB> + Send,
49    E: for<'r> FromRow<'r, <DB as Database>::Row> + Send + Unpin,
50{
51    let query = QueryBuilder::select(dialect, E::table_name())
52        .columns(&column_names::<E>())
53        .build();
54
55    query.fetch_all::<DB, E, _>(executor).await
56}
57
58/// Build an INSERT statement for `entity`, excluding the primary key column.
59pub(crate) fn build_insert_query<E, D>(dialect: &D, entity: &E) -> BuiltQuery
60where
61    E: Entity,
62    D: Dialect,
63{
64    let cols = E::columns();
65    let id_col = E::id_column();
66    let values = entity.values();
67
68    let mut builder = QueryBuilder::insert(dialect, E::table_name());
69    for (col, value) in cols.iter().zip(values.iter()) {
70        if col.name() != id_col {
71            builder = builder.set(col.name(), value.clone());
72        }
73    }
74
75    if dialect.supports_returning() {
76        builder.returning(&column_names::<E>()).build()
77    } else {
78        builder.build()
79    }
80}
81
82/// Build an UPDATE statement for `entity`, excluding the primary key from SET.
83pub(crate) fn build_update_query<E, D>(dialect: &D, entity: &E) -> BuiltQuery
84where
85    E: Entity,
86    D: Dialect,
87{
88    let cols = E::columns();
89    let id_col = E::id_column();
90    let values = entity.values();
91    let id = entity.id();
92
93    let mut builder = QueryBuilder::update(dialect, E::table_name());
94    for (col, value) in cols.iter().zip(values.iter()) {
95        if col.name() != id_col {
96            builder = builder.set(col.name(), value.clone());
97        }
98    }
99    builder = builder.filter(Filter::new(id_col, Op::Eq, id));
100
101    if dialect.supports_returning() {
102        builder.returning(&column_names::<E>()).build()
103    } else {
104        builder.build()
105    }
106}
107
108/// Build a SELECT for the row most recently inserted on engines without RETURNING.
109#[cfg(test)]
110pub(crate) fn build_select_last_inserted_query<E, D>(dialect: &D) -> BuiltQuery
111where
112    E: Entity,
113    D: Dialect,
114{
115    let table = dialect.quote_identifier(E::table_name());
116    let id_col = dialect.quote_identifier(E::id_column());
117    let columns = E::columns()
118        .iter()
119        .map(|c| dialect.quote_identifier(c.name()))
120        .collect::<Vec<_>>()
121        .join(", ");
122
123    BuiltQuery {
124        sql: format!("SELECT {columns} FROM {table} WHERE {id_col} = LAST_INSERT_ID()"),
125        values: Vec::new(),
126    }
127}
128
129/// Insert a new entity and return it.
130///
131/// The id column is excluded from the INSERT statement so that the database
132/// can assign an autoincrement primary key. On engines without `RETURNING`,
133/// the inserted row is re-selected via `LAST_INSERT_ID()`.
134pub async fn insert<'e, E, DB, D, Ex>(
135    dialect: &D,
136    executor: Ex,
137    entity: &E,
138) -> Result<E, FletchError>
139where
140    E: Entity,
141    D: Dialect,
142    DB: Database + DatabaseBindable,
143    Ex: Executor<'e, Database = DB> + Send,
144    E: for<'r> FromRow<'r, <DB as Database>::Row> + Send + Unpin,
145{
146    let query = build_insert_query::<E, D>(dialect, entity);
147
148    if dialect.supports_returning() {
149        return query.fetch_one::<DB, E, _>(executor).await;
150    }
151
152    query.execute::<DB, _>(executor).await?;
153    Err(FletchError::Configuration(
154        "insert on engines without RETURNING requires a re-select executor pass; use Pool::insert or Transaction::insert".into(),
155    ))
156}
157
158/// Update an existing entity (matched by id) and return the updated row.
159pub async fn update<'e, E, DB, D, Ex>(
160    dialect: &D,
161    executor: Ex,
162    entity: &E,
163) -> Result<E, FletchError>
164where
165    E: Entity,
166    D: Dialect,
167    DB: Database + DatabaseBindable,
168    Ex: Executor<'e, Database = DB> + Send,
169    E: for<'r> FromRow<'r, <DB as Database>::Row> + Send + Unpin,
170{
171    let query = build_update_query::<E, D>(dialect, entity);
172
173    if dialect.supports_returning() {
174        return query.fetch_one::<DB, E, _>(executor).await;
175    }
176
177    let id = entity.id();
178    let affected = query.execute::<DB, _>(executor).await?;
179    if affected == 0 {
180        return Err(FletchError::NotFound);
181    }
182
183    let _ = id;
184    Err(FletchError::Configuration(
185        "update on engines without RETURNING requires a re-select executor pass; use Pool::update or Transaction::update".into(),
186    ))
187}
188
189/// Delete an entity by its primary key.
190pub async fn delete<'e, E, DB, D, Ex>(
191    dialect: &D,
192    executor: Ex,
193    id: E::Id,
194) -> Result<(), FletchError>
195where
196    E: Entity,
197    D: Dialect,
198    DB: Database + DatabaseBindable,
199    Ex: Executor<'e, Database = DB> + Send,
200{
201    let query = QueryBuilder::delete(dialect, E::table_name())
202        .filter(Filter::new(E::id_column(), Op::Eq, id))
203        .build();
204
205    let affected = query.execute::<DB, _>(executor).await?;
206    if affected == 0 {
207        return Err(FletchError::NotFound);
208    }
209    Ok(())
210}
211
212/// Execute a raw SQL query and return all matching rows.
213pub async fn fetch_all<'e, T, DB, Ex>(
214    sql: &str,
215    params: &[Value],
216    executor: Ex,
217) -> Result<Vec<T>, FletchError>
218where
219    DB: Database + DatabaseBindable,
220    Ex: Executor<'e, Database = DB> + Send,
221    T: for<'r> FromRow<'r, <DB as Database>::Row> + Send + Unpin,
222{
223    BuiltQuery {
224        sql: sql.to_owned(),
225        values: params.to_vec(),
226    }
227    .fetch_all::<DB, T, _>(executor)
228    .await
229}
230
231fn column_names<E: Entity>() -> Vec<&'static str> {
232    E::columns().iter().map(Column::name).collect()
233}
234
235// ---------------------------------------------------------------------------
236// Pool and Transaction passthrough
237// ---------------------------------------------------------------------------
238
239#[cfg(any(feature = "sqlite", feature = "postgres", feature = "mysql"))]
240macro_rules! impl_crud {
241    ($db:ty, $dialect:expr, $fetch_one:path, $fetch_all:path, $execute:path) => {
242        impl Pool<$db> {
243            /// Find an entity by its primary key.
244            pub async fn find_by_id<E>(&self, id: E::Id) -> Result<E, FletchError>
245            where
246                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
247            {
248                let query = QueryBuilder::select($dialect, E::table_name())
249                    .columns(&column_names::<E>())
250                    .filter(Filter::new(E::id_column(), Op::Eq, id))
251                    .build();
252                ($fetch_one)(query, self.inner()).await
253            }
254
255            /// Fetch all entities of this type.
256            pub async fn find_all<E>(&self) -> Result<Vec<E>, FletchError>
257            where
258                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
259            {
260                let query = QueryBuilder::select($dialect, E::table_name())
261                    .columns(&column_names::<E>())
262                    .build();
263                ($fetch_all)(query, self.inner()).await
264            }
265
266            /// Insert a new entity and return it.
267            pub async fn insert<E>(&self, entity: &E) -> Result<E, FletchError>
268            where
269                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
270            {
271                let query = build_insert_query::<E, _>($dialect, entity);
272                if $dialect.supports_returning() {
273                    ($fetch_one)(query, self.inner()).await
274                } else {
275                    let outcome = ($execute)(query, self.inner()).await?;
276                    let insert_id = outcome.last_insert_id.ok_or_else(|| {
277                        FletchError::Configuration("insert did not return a last insert id".into())
278                    })?;
279                    let insert_id = i64::try_from(insert_id).map_err(|_| {
280                        FletchError::Configuration("insert last insert id exceeds i64 range".into())
281                    })?;
282                    let reselect = QueryBuilder::select($dialect, E::table_name())
283                        .columns(&column_names::<E>())
284                        .filter(Filter::new(E::id_column(), Op::Eq, Value::I64(insert_id)))
285                        .build();
286                    ($fetch_one)(reselect, self.inner()).await
287                }
288            }
289
290            /// Update an existing entity and return the updated row.
291            pub async fn update<E>(&self, entity: &E) -> Result<E, FletchError>
292            where
293                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
294            {
295                let query = build_update_query::<E, _>($dialect, entity);
296                if $dialect.supports_returning() {
297                    ($fetch_one)(query, self.inner()).await
298                } else {
299                    let id = entity.id();
300                    let affected = ($execute)(query, self.inner()).await?.rows_affected;
301                    if affected == 0 {
302                        return Err(FletchError::NotFound);
303                    }
304                    self.find_by_id::<E>(id).await
305                }
306            }
307
308            /// Delete an entity by its primary key.
309            pub async fn delete<E>(&self, id: E::Id) -> Result<(), FletchError>
310            where
311                E: Entity,
312            {
313                let query = QueryBuilder::delete($dialect, E::table_name())
314                    .filter(Filter::new(E::id_column(), Op::Eq, id))
315                    .build();
316                let affected = ($execute)(query, self.inner()).await?.rows_affected;
317                if affected == 0 {
318                    return Err(FletchError::NotFound);
319                }
320                Ok(())
321            }
322
323            /// Execute a raw SQL query and return all matching rows.
324            pub async fn fetch_all<T>(
325                &self,
326                sql: &str,
327                params: &[Value],
328            ) -> Result<Vec<T>, FletchError>
329            where
330                T: for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
331            {
332                ($fetch_all)(
333                    BuiltQuery {
334                        sql: sql.to_owned(),
335                        values: params.to_vec(),
336                    },
337                    self.inner(),
338                )
339                .await
340            }
341        }
342
343        impl<'a> Transaction<'a, $db> {
344            /// Find an entity by its primary key.
345            pub async fn find_by_id<E>(&mut self, id: E::Id) -> Result<E, FletchError>
346            where
347                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
348            {
349                let query = QueryBuilder::select($dialect, E::table_name())
350                    .columns(&column_names::<E>())
351                    .filter(Filter::new(E::id_column(), Op::Eq, id))
352                    .build();
353                ($fetch_one)(query, &mut **self.inner_mut()).await
354            }
355
356            /// Fetch all entities of this type.
357            pub async fn find_all<E>(&mut self) -> Result<Vec<E>, FletchError>
358            where
359                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
360            {
361                let query = QueryBuilder::select($dialect, E::table_name())
362                    .columns(&column_names::<E>())
363                    .build();
364                ($fetch_all)(query, &mut **self.inner_mut()).await
365            }
366
367            /// Insert a new entity and return it.
368            pub async fn insert<E>(&mut self, entity: &E) -> Result<E, FletchError>
369            where
370                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
371            {
372                let query = build_insert_query::<E, _>($dialect, entity);
373                if $dialect.supports_returning() {
374                    ($fetch_one)(query, &mut **self.inner_mut()).await
375                } else {
376                    let outcome = ($execute)(query, &mut **self.inner_mut()).await?;
377                    let insert_id = outcome.last_insert_id.ok_or_else(|| {
378                        FletchError::Configuration("insert did not return a last insert id".into())
379                    })?;
380                    let insert_id = i64::try_from(insert_id).map_err(|_| {
381                        FletchError::Configuration("insert last insert id exceeds i64 range".into())
382                    })?;
383                    let reselect = QueryBuilder::select($dialect, E::table_name())
384                        .columns(&column_names::<E>())
385                        .filter(Filter::new(E::id_column(), Op::Eq, Value::I64(insert_id)))
386                        .build();
387                    ($fetch_one)(reselect, &mut **self.inner_mut()).await
388                }
389            }
390
391            /// Update an existing entity and return the updated row.
392            pub async fn update<E>(&mut self, entity: &E) -> Result<E, FletchError>
393            where
394                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
395            {
396                let query = build_update_query::<E, _>($dialect, entity);
397                if $dialect.supports_returning() {
398                    ($fetch_one)(query, &mut **self.inner_mut()).await
399                } else {
400                    let id = entity.id();
401                    let affected = ($execute)(query, &mut **self.inner_mut())
402                        .await?
403                        .rows_affected;
404                    if affected == 0 {
405                        return Err(FletchError::NotFound);
406                    }
407                    self.find_by_id::<E>(id).await
408                }
409            }
410
411            /// Delete an entity by its primary key.
412            pub async fn delete<E>(&mut self, id: E::Id) -> Result<(), FletchError>
413            where
414                E: Entity,
415            {
416                let query = QueryBuilder::delete($dialect, E::table_name())
417                    .filter(Filter::new(E::id_column(), Op::Eq, id))
418                    .build();
419                let affected = ($execute)(query, &mut **self.inner_mut())
420                    .await?
421                    .rows_affected;
422                if affected == 0 {
423                    return Err(FletchError::NotFound);
424                }
425                Ok(())
426            }
427
428            /// Execute a raw SQL query and return all matching rows.
429            pub async fn fetch_all<T>(
430                &mut self,
431                sql: &str,
432                params: &[Value],
433            ) -> Result<Vec<T>, FletchError>
434            where
435                T: for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
436            {
437                ($fetch_all)(
438                    BuiltQuery {
439                        sql: sql.to_owned(),
440                        values: params.to_vec(),
441                    },
442                    &mut **self.inner_mut(),
443                )
444                .await
445            }
446        }
447    };
448}
449
450#[cfg(feature = "sqlite")]
451impl_crud!(
452    sqlx::Sqlite,
453    &crate::dialect::SqliteDialect,
454    crate::bind::sqlite_bind::fetch_one_row,
455    crate::bind::sqlite_bind::fetch_all_rows,
456    crate::bind::sqlite_bind::execute_query
457);
458
459#[cfg(feature = "postgres")]
460impl_crud!(
461    sqlx::Postgres,
462    &crate::dialect::PostgresDialect,
463    crate::bind::postgres_bind::fetch_one_row,
464    crate::bind::postgres_bind::fetch_all_rows,
465    crate::bind::postgres_bind::execute_query
466);
467
468#[cfg(feature = "mysql")]
469impl_crud!(
470    sqlx::MySql,
471    &crate::dialect::MySqlDialect,
472    crate::bind::mysql_bind::fetch_one_row,
473    crate::bind::mysql_bind::fetch_all_rows,
474    crate::bind::mysql_bind::execute_query
475);
476
477#[cfg(test)]
478mod tests {
479    use super::*;
480    use crate::column::{Column, ColumnType};
481
482    #[derive(Debug, Clone, PartialEq, sqlx::FromRow)]
483    struct User {
484        id: i64,
485        name: String,
486    }
487
488    impl Entity for User {
489        type Id = i64;
490
491        fn table_name() -> &'static str {
492            "users"
493        }
494
495        fn id_column() -> &'static str {
496            "id"
497        }
498
499        fn columns() -> &'static [Column] {
500            static COLS: [Column; 2] = [
501                Column::new("id", ColumnType::Integer),
502                Column::new("name", ColumnType::Text),
503            ];
504            &COLS
505        }
506
507        fn id(&self) -> Self::Id {
508            self.id
509        }
510
511        fn values(&self) -> Vec<Value> {
512            vec![Value::I64(self.id), Value::Text(self.name.clone())]
513        }
514    }
515
516    #[cfg(feature = "sqlite")]
517    async fn setup_pool() -> Pool<sqlx::Sqlite> {
518        let pool = Pool::<sqlx::Sqlite>::connect("sqlite::memory:")
519            .await
520            .unwrap();
521        sqlx::query("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL)")
522            .execute(pool.inner())
523            .await
524            .unwrap();
525        pool
526    }
527
528    #[cfg(feature = "sqlite")]
529    #[tokio::test]
530    async fn insert_and_find_by_id() {
531        let pool = setup_pool().await;
532        let user = User {
533            id: 0,
534            name: "Alice".into(),
535        };
536        let inserted = pool.insert(&user).await.unwrap();
537        assert_ne!(inserted.id, 0);
538        assert_eq!(inserted.name, "Alice");
539
540        let found = pool.find_by_id::<User>(inserted.id).await.unwrap();
541        assert_eq!(found, inserted);
542    }
543
544    #[cfg(feature = "sqlite")]
545    #[tokio::test]
546    async fn find_by_id_not_found() {
547        let pool = setup_pool().await;
548        let err = pool.find_by_id::<User>(999).await.unwrap_err();
549        assert!(matches!(err, FletchError::NotFound));
550    }
551
552    #[cfg(feature = "sqlite")]
553    #[tokio::test]
554    async fn find_all_empty() {
555        let pool = setup_pool().await;
556        let users = pool.find_all::<User>().await.unwrap();
557        assert!(users.is_empty());
558    }
559
560    #[cfg(feature = "sqlite")]
561    #[tokio::test]
562    async fn find_all_with_data() {
563        let pool = setup_pool().await;
564        pool.insert(&User {
565            id: 0,
566            name: "Alice".into(),
567        })
568        .await
569        .unwrap();
570        pool.insert(&User {
571            id: 0,
572            name: "Bob".into(),
573        })
574        .await
575        .unwrap();
576
577        let users = pool.find_all::<User>().await.unwrap();
578        assert_eq!(users.len(), 2);
579    }
580
581    #[cfg(feature = "sqlite")]
582    #[tokio::test]
583    async fn update_entity() {
584        let pool = setup_pool().await;
585        let inserted = pool
586            .insert(&User {
587                id: 0,
588                name: "Alice".into(),
589            })
590            .await
591            .unwrap();
592
593        let updated = pool
594            .update(&User {
595                id: inserted.id,
596                name: "Alicia".into(),
597            })
598            .await
599            .unwrap();
600        assert_eq!(updated.name, "Alicia");
601
602        let found = pool.find_by_id::<User>(inserted.id).await.unwrap();
603        assert_eq!(found.name, "Alicia");
604    }
605
606    #[cfg(feature = "sqlite")]
607    #[tokio::test]
608    async fn delete_entity() {
609        let pool = setup_pool().await;
610        let inserted = pool
611            .insert(&User {
612                id: 0,
613                name: "Alice".into(),
614            })
615            .await
616            .unwrap();
617
618        pool.delete::<User>(inserted.id).await.unwrap();
619
620        let err = pool.find_by_id::<User>(inserted.id).await.unwrap_err();
621        assert!(matches!(err, FletchError::NotFound));
622    }
623
624    #[cfg(feature = "sqlite")]
625    #[tokio::test]
626    async fn delete_not_found() {
627        let pool = setup_pool().await;
628        let err = pool.delete::<User>(999).await.unwrap_err();
629        assert!(matches!(err, FletchError::NotFound));
630    }
631
632    #[cfg(feature = "sqlite")]
633    #[tokio::test]
634    async fn fetch_all_raw_sql() {
635        let pool = setup_pool().await;
636        pool.insert(&User {
637            id: 0,
638            name: "Alice".into(),
639        })
640        .await
641        .unwrap();
642        pool.insert(&User {
643            id: 0,
644            name: "Bob".into(),
645        })
646        .await
647        .unwrap();
648
649        let users: Vec<User> = pool
650            .fetch_all(
651                "SELECT id, name FROM users WHERE name LIKE ?",
652                &[Value::Text("%li%".into())],
653            )
654            .await
655            .unwrap();
656        assert_eq!(users.len(), 1);
657        assert_eq!(users[0].name, "Alice");
658    }
659
660    #[cfg(feature = "sqlite")]
661    #[tokio::test]
662    async fn insert_multiple_gets_unique_ids() {
663        let pool = setup_pool().await;
664        let a = pool
665            .insert(&User {
666                id: 0,
667                name: "Alice".into(),
668            })
669            .await
670            .unwrap();
671        let b = pool
672            .insert(&User {
673                id: 0,
674                name: "Bob".into(),
675            })
676            .await
677            .unwrap();
678        let c = pool
679            .insert(&User {
680                id: 0,
681                name: "Carol".into(),
682            })
683            .await
684            .unwrap();
685
686        assert_ne!(a.id, 0);
687        assert_ne!(b.id, 0);
688        assert_ne!(c.id, 0);
689        assert_ne!(a.id, b.id);
690        assert_ne!(b.id, c.id);
691        assert_ne!(a.id, c.id);
692
693        let all = pool.find_all::<User>().await.unwrap();
694        assert_eq!(all.len(), 3);
695    }
696
697    #[cfg(feature = "sqlite")]
698    #[tokio::test]
699    async fn crud_in_transaction() {
700        let pool = setup_pool().await;
701
702        let mut tx = pool.begin().await.unwrap();
703        tx.insert(&User {
704            id: 0,
705            name: "Alice".into(),
706        })
707        .await
708        .unwrap();
709        tx.commit().await.unwrap();
710
711        let users = pool.find_all::<User>().await.unwrap();
712        assert_eq!(users.len(), 1);
713        assert_eq!(users[0].name, "Alice");
714    }
715
716    #[cfg(feature = "postgres")]
717    mod postgres {
718        use super::*;
719
720        static SCHEMA_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
721
722        #[derive(Debug, Clone, PartialEq, sqlx::FromRow)]
723        struct PgUser {
724            id: i64,
725            name: String,
726        }
727
728        impl Entity for PgUser {
729            type Id = i64;
730
731            fn table_name() -> &'static str {
732                "fletch_users"
733            }
734
735            fn id_column() -> &'static str {
736                "id"
737            }
738
739            fn columns() -> &'static [Column] {
740                static COLS: [Column; 2] = [
741                    Column::new("id", ColumnType::Integer),
742                    Column::new("name", ColumnType::Text),
743                ];
744                &COLS
745            }
746
747            fn id(&self) -> Self::Id {
748                self.id
749            }
750
751            fn values(&self) -> Vec<Value> {
752                vec![Value::I64(self.id), Value::Text(self.name.clone())]
753            }
754        }
755
756        fn database_url() -> String {
757            std::env::var("DATABASE_URL")
758                .unwrap_or_else(|_| "postgres://postgres:postgres@127.0.0.1:5432/aro_test".into())
759        }
760
761        async fn setup_pool() -> (Pool<sqlx::Postgres>, tokio::sync::MutexGuard<'static, ()>) {
762            let lock = SCHEMA_LOCK.lock().await;
763            let pool = Pool::<sqlx::Postgres>::connect(&database_url())
764                .await
765                .unwrap();
766            sqlx::query("DROP TABLE IF EXISTS fletch_users")
767                .execute(pool.inner())
768                .await
769                .unwrap();
770            sqlx::query("CREATE TABLE fletch_users (id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL)")
771                .execute(pool.inner())
772                .await
773                .unwrap();
774            (pool, lock)
775        }
776
777        #[tokio::test]
778        #[ignore = "requires DATABASE_URL and running Postgres"]
779        async fn insert_and_find_by_id() {
780            let (pool, _lock) = setup_pool().await;
781            let user = PgUser {
782                id: 0,
783                name: "Alice".into(),
784            };
785            let inserted = pool.insert(&user).await.unwrap();
786            assert_ne!(inserted.id, 0);
787            assert_eq!(inserted.name, "Alice");
788
789            let found = pool.find_by_id::<PgUser>(inserted.id).await.unwrap();
790            assert_eq!(found, inserted);
791        }
792
793        #[tokio::test]
794        #[ignore = "requires DATABASE_URL and running Postgres"]
795        async fn find_all_with_data() {
796            let (pool, _lock) = setup_pool().await;
797            pool.insert(&PgUser {
798                id: 0,
799                name: "Alice".into(),
800            })
801            .await
802            .unwrap();
803            pool.insert(&PgUser {
804                id: 0,
805                name: "Bob".into(),
806            })
807            .await
808            .unwrap();
809
810            let users = pool.find_all::<PgUser>().await.unwrap();
811            assert_eq!(users.len(), 2);
812        }
813
814        #[tokio::test]
815        #[ignore = "requires DATABASE_URL and running Postgres"]
816        async fn update_entity() {
817            let (pool, _lock) = setup_pool().await;
818            let inserted = pool
819                .insert(&PgUser {
820                    id: 0,
821                    name: "Alice".into(),
822                })
823                .await
824                .unwrap();
825
826            let updated = pool
827                .update(&PgUser {
828                    id: inserted.id,
829                    name: "Alicia".into(),
830                })
831                .await
832                .unwrap();
833            assert_eq!(updated.name, "Alicia");
834
835            let found = pool.find_by_id::<PgUser>(inserted.id).await.unwrap();
836            assert_eq!(found.name, "Alicia");
837        }
838
839        #[tokio::test]
840        #[ignore = "requires DATABASE_URL and running Postgres"]
841        async fn delete_entity() {
842            let (pool, _lock) = setup_pool().await;
843            let inserted = pool
844                .insert(&PgUser {
845                    id: 0,
846                    name: "Alice".into(),
847                })
848                .await
849                .unwrap();
850
851            pool.delete::<PgUser>(inserted.id).await.unwrap();
852
853            let err = pool.find_by_id::<PgUser>(inserted.id).await.unwrap_err();
854            assert!(matches!(err, FletchError::NotFound));
855        }
856
857        #[tokio::test]
858        #[ignore = "requires DATABASE_URL and running Postgres"]
859        async fn delete_not_found() {
860            let (pool, _lock) = setup_pool().await;
861            let err = pool.delete::<PgUser>(999).await.unwrap_err();
862            assert!(matches!(err, FletchError::NotFound));
863        }
864
865        #[tokio::test]
866        #[ignore = "requires DATABASE_URL and running Postgres"]
867        async fn fetch_all_raw_sql() {
868            let (pool, _lock) = setup_pool().await;
869            pool.insert(&PgUser {
870                id: 0,
871                name: "Alice".into(),
872            })
873            .await
874            .unwrap();
875            pool.insert(&PgUser {
876                id: 0,
877                name: "Bob".into(),
878            })
879            .await
880            .unwrap();
881
882            let users: Vec<PgUser> = pool
883                .fetch_all(
884                    "SELECT id, name FROM fletch_users WHERE name LIKE $1",
885                    &[Value::Text("%li%".into())],
886                )
887                .await
888                .unwrap();
889            assert_eq!(users.len(), 1);
890            assert_eq!(users[0].name, "Alice");
891        }
892
893        #[tokio::test]
894        #[ignore = "requires DATABASE_URL and running Postgres"]
895        async fn crud_in_transaction() {
896            let (pool, _lock) = setup_pool().await;
897
898            let mut tx = pool.begin().await.unwrap();
899            tx.insert(&PgUser {
900                id: 0,
901                name: "Alice".into(),
902            })
903            .await
904            .unwrap();
905            tx.commit().await.unwrap();
906
907            let users = pool.find_all::<PgUser>().await.unwrap();
908            assert_eq!(users.len(), 1);
909            assert_eq!(users[0].name, "Alice");
910        }
911    }
912
913    #[cfg(feature = "mysql")]
914    mod mysql {
915        use super::*;
916
917        static SCHEMA_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
918
919        #[derive(Debug, Clone, PartialEq, sqlx::FromRow)]
920        struct MyUser {
921            id: i64,
922            name: String,
923        }
924
925        impl Entity for MyUser {
926            type Id = i64;
927
928            fn table_name() -> &'static str {
929                "fletch_users"
930            }
931
932            fn id_column() -> &'static str {
933                "id"
934            }
935
936            fn columns() -> &'static [Column] {
937                static COLS: [Column; 2] = [
938                    Column::new("id", ColumnType::Integer),
939                    Column::new("name", ColumnType::Text),
940                ];
941                &COLS
942            }
943
944            fn id(&self) -> Self::Id {
945                self.id
946            }
947
948            fn values(&self) -> Vec<Value> {
949                vec![Value::I64(self.id), Value::Text(self.name.clone())]
950            }
951        }
952
953        fn database_url() -> String {
954            std::env::var("DATABASE_URL")
955                .unwrap_or_else(|_| "mysql://root:mysql@127.0.0.1:3306/aro_test".into())
956        }
957
958        async fn setup_pool() -> (Pool<sqlx::MySql>, tokio::sync::MutexGuard<'static, ()>) {
959            let lock = SCHEMA_LOCK.lock().await;
960            let pool = Pool::<sqlx::MySql>::connect(&database_url()).await.unwrap();
961            sqlx::query("DROP TABLE IF EXISTS fletch_users")
962                .execute(pool.inner())
963                .await
964                .unwrap();
965            sqlx::query(
966                "CREATE TABLE fletch_users (id BIGINT AUTO_INCREMENT PRIMARY KEY, name TEXT NOT NULL)",
967            )
968            .execute(pool.inner())
969            .await
970            .unwrap();
971            (pool, lock)
972        }
973
974        #[tokio::test]
975        #[ignore = "requires DATABASE_URL and running MySQL"]
976        async fn mysql_insert_without_returning() {
977            let (pool, _lock) = setup_pool().await;
978            let inserted = pool
979                .insert(&MyUser {
980                    id: 0,
981                    name: "Alice".into(),
982                })
983                .await
984                .unwrap();
985            assert!(inserted.id > 0);
986            assert_eq!(inserted.name, "Alice");
987
988            let updated = pool
989                .update(&MyUser {
990                    id: inserted.id,
991                    name: "Alicia".into(),
992                })
993                .await
994                .unwrap();
995            assert_eq!(updated.name, "Alicia");
996
997            pool.delete::<MyUser>(inserted.id).await.unwrap();
998        }
999
1000        #[tokio::test]
1001        #[ignore = "requires DATABASE_URL and running MySQL"]
1002        async fn find_all_with_data() {
1003            let (pool, _lock) = setup_pool().await;
1004            pool.insert(&MyUser {
1005                id: 0,
1006                name: "Alice".into(),
1007            })
1008            .await
1009            .unwrap();
1010            pool.insert(&MyUser {
1011                id: 0,
1012                name: "Bob".into(),
1013            })
1014            .await
1015            .unwrap();
1016
1017            let users = pool.find_all::<MyUser>().await.unwrap();
1018            assert_eq!(users.len(), 2);
1019        }
1020
1021        #[tokio::test]
1022        #[ignore = "requires DATABASE_URL and running MySQL"]
1023        async fn find_by_id_not_found() {
1024            let (pool, _lock) = setup_pool().await;
1025            let err = pool.find_by_id::<MyUser>(999).await.unwrap_err();
1026            assert!(matches!(err, FletchError::NotFound));
1027        }
1028
1029        #[tokio::test]
1030        #[ignore = "requires DATABASE_URL and running MySQL"]
1031        async fn delete_not_found() {
1032            let (pool, _lock) = setup_pool().await;
1033            let err = pool.delete::<MyUser>(999).await.unwrap_err();
1034            assert!(matches!(err, FletchError::NotFound));
1035        }
1036
1037        #[tokio::test]
1038        #[ignore = "requires DATABASE_URL and running MySQL"]
1039        async fn fetch_all_raw_sql() {
1040            let (pool, _lock) = setup_pool().await;
1041            pool.insert(&MyUser {
1042                id: 0,
1043                name: "Alice".into(),
1044            })
1045            .await
1046            .unwrap();
1047            pool.insert(&MyUser {
1048                id: 0,
1049                name: "Bob".into(),
1050            })
1051            .await
1052            .unwrap();
1053
1054            let users: Vec<MyUser> = pool
1055                .fetch_all(
1056                    "SELECT id, name FROM fletch_users WHERE name LIKE ?",
1057                    &[Value::Text("%li%".into())],
1058                )
1059                .await
1060                .unwrap();
1061            assert_eq!(users.len(), 1);
1062            assert_eq!(users[0].name, "Alice");
1063        }
1064
1065        #[tokio::test]
1066        #[ignore = "requires DATABASE_URL and running MySQL"]
1067        async fn crud_in_transaction() {
1068            let (pool, _lock) = setup_pool().await;
1069
1070            let mut tx = pool.begin().await.unwrap();
1071            tx.insert(&MyUser {
1072                id: 0,
1073                name: "Alice".into(),
1074            })
1075            .await
1076            .unwrap();
1077            tx.commit().await.unwrap();
1078
1079            let users = pool.find_all::<MyUser>().await.unwrap();
1080            assert_eq!(users.len(), 1);
1081            assert_eq!(users[0].name, "Alice");
1082        }
1083
1084        #[tokio::test]
1085        #[ignore = "requires DATABASE_URL and running MySQL"]
1086        async fn update_in_transaction() {
1087            let (pool, _lock) = setup_pool().await;
1088            let inserted = pool
1089                .insert(&MyUser {
1090                    id: 0,
1091                    name: "Alice".into(),
1092                })
1093                .await
1094                .unwrap();
1095
1096            let mut tx = pool.begin().await.unwrap();
1097            let updated = tx
1098                .update(&MyUser {
1099                    id: inserted.id,
1100                    name: "Alicia".into(),
1101                })
1102                .await
1103                .unwrap();
1104            assert_eq!(updated.name, "Alicia");
1105            tx.commit().await.unwrap();
1106
1107            let found = pool.find_by_id::<MyUser>(inserted.id).await.unwrap();
1108            assert_eq!(found.name, "Alicia");
1109        }
1110
1111        #[tokio::test]
1112        #[ignore = "requires DATABASE_URL and running MySQL"]
1113        async fn insert_multiple_gets_unique_ids() {
1114            let (pool, _lock) = setup_pool().await;
1115            let a = pool
1116                .insert(&MyUser {
1117                    id: 0,
1118                    name: "Alice".into(),
1119                })
1120                .await
1121                .unwrap();
1122            let b = pool
1123                .insert(&MyUser {
1124                    id: 0,
1125                    name: "Bob".into(),
1126                })
1127                .await
1128                .unwrap();
1129            let c = pool
1130                .insert(&MyUser {
1131                    id: 0,
1132                    name: "Carol".into(),
1133                })
1134                .await
1135                .unwrap();
1136
1137            assert_ne!(a.id, 0);
1138            assert_ne!(b.id, 0);
1139            assert_ne!(c.id, 0);
1140            assert_ne!(a.id, b.id);
1141            assert_ne!(b.id, c.id);
1142            assert_ne!(a.id, c.id);
1143
1144            let all = pool.find_all::<MyUser>().await.unwrap();
1145            assert_eq!(all.len(), 3);
1146        }
1147    }
1148
1149    #[test]
1150    fn build_insert_query_sqlite_uses_returning() {
1151        let dialect = crate::dialect::SqliteDialect;
1152        let user = User {
1153            id: 0,
1154            name: "Alice".into(),
1155        };
1156        let query = build_insert_query::<User, _>(&dialect, &user);
1157        assert!(query.sql.contains("RETURNING"));
1158        assert!(query.sql.contains("(\"name\")"));
1159        assert!(!query.sql.contains("(\"id\""));
1160    }
1161
1162    #[test]
1163    fn build_insert_query_mysql_omits_returning() {
1164        let dialect = crate::dialect::MySqlDialect;
1165        let user = User {
1166            id: 0,
1167            name: "Alice".into(),
1168        };
1169        let query = build_insert_query::<User, _>(&dialect, &user);
1170        assert!(!query.sql.contains("RETURNING"));
1171        assert!(query.sql.contains("`name`"));
1172    }
1173
1174    #[test]
1175    fn build_select_last_inserted_mysql() {
1176        let dialect = crate::dialect::MySqlDialect;
1177        let query = build_select_last_inserted_query::<User, _>(&dialect);
1178        assert!(query.sql.contains("LAST_INSERT_ID()"));
1179        assert!(query.values.is_empty());
1180    }
1181}
fletch_orm/ops.rs

fletch_orm/
ops.rs
