floz_orm/db/
executor.rs

1//! `Executor` trait and `bind_params!` macro.
2//!
3//! The `Executor` trait is the core abstraction for database operations.
4//! Both `Db` (connection pool) and `Tx` (transaction) implement it,
5//! so DAO methods accept `&impl Executor` generically.
6
7use crate::error::FlozError;
8use crate::value::Value;
9use super::row_bound::FlozRowBound;
10use super::pool::{Db, DbInner};
11use super::tx::{Tx, TxInner};
12
13// ═══════════════════════════════════════════════════════════════
14// Executor Trait
15// ═══════════════════════════════════════════════════════════════
16
17/// The core database execution trait.
18///
19/// All floz DAO and DSL methods accept `&impl Executor`, allowing
20/// seamless switching between connection pools, transactions, and mocks.
21///
22/// ```ignore
23/// // Both work:
24/// let users = User::all(&db).await?;
25/// let users = User::all(&tx).await?;
26/// ```
27pub trait Executor {
28    /// Execute a raw SQL statement, returning the number of affected rows.
29    fn execute_raw(
30        &self,
31        sql: &str,
32        params: Vec<Value>,
33    ) -> impl std::future::Future<Output = Result<u64, FlozError>> + Send;
34
35    /// Fetch all matching rows.
36    fn fetch_all<T>(
37        &self,
38        sql: &str,
39        params: Vec<Value>,
40    ) -> impl std::future::Future<Output = Result<Vec<T>, FlozError>> + Send
41    where
42        T: Send + Unpin + FlozRowBound;
43
44    /// Fetch exactly one row. Returns `FlozError::NotFound` if no rows match.
45    fn fetch_one<T>(
46        &self,
47        sql: &str,
48        params: Vec<Value>,
49    ) -> impl std::future::Future<Output = Result<T, FlozError>> + Send
50    where
51        T: Send + Unpin + FlozRowBound;
52
53    /// Fetch zero or one row.
54    fn fetch_optional<T>(
55        &self,
56        sql: &str,
57        params: Vec<Value>,
58    ) -> impl std::future::Future<Output = Result<Option<T>, FlozError>> + Send
59    where
60        T: Send + Unpin + FlozRowBound;
61}
62
63// ═══════════════════════════════════════════════════════════════
64// bind_params! macro
65// ═══════════════════════════════════════════════════════════════
66
67/// Bind a `Vec<Value>` to a `sqlx::query()` builder.
68///
69/// Each `Value` variant is pattern-matched to call `.bind()` with the
70/// concrete type. Works for both Postgres and SQLite.
71macro_rules! bind_params {
72    ($query:expr, $params:expr) => {{
73        let mut q = $query;
74        for param in $params {
75            q = match param {
76                // Non-nullable
77                Value::Short(v) => q.bind(v),
78                Value::Int(v) => q.bind(v),
79                Value::BigInt(v) => q.bind(v),
80                Value::Real(v) => q.bind(v),
81                Value::Double(v) => q.bind(v),
82                Value::Bool(v) => q.bind(v),
83                Value::String(v) => q.bind(v),
84                Value::Bytes(v) => q.bind(v),
85                Value::Uuid(v) => q.bind(v.to_string()),
86                Value::DateTime(v) => q.bind(v),
87                Value::NaiveDateTime(v) => q.bind(v),
88                Value::NaiveDate(v) => q.bind(v),
89                Value::NaiveTime(v) => q.bind(v),
90                Value::Json(v) => q.bind(v.to_string()),
91                Value::Jsonb(v) => q.bind(v.to_string()),
92                // Nullable
93                Value::OptionShort(v) => q.bind(v),
94                Value::OptionInt(v) => q.bind(v),
95                Value::OptionBigInt(v) => q.bind(v),
96                Value::OptionReal(v) => q.bind(v),
97                Value::OptionDouble(v) => q.bind(v),
98                Value::OptionBool(v) => q.bind(v),
99                Value::OptionString(v) => q.bind(v),
100                Value::OptionBytes(v) => q.bind(v),
101                Value::OptionUuid(v) => q.bind(v.map(|u| u.to_string())),
102                Value::OptionDateTime(v) => q.bind(v),
103                Value::OptionNaiveDateTime(v) => q.bind(v),
104                Value::OptionNaiveDate(v) => q.bind(v),
105                Value::OptionNaiveTime(v) => q.bind(v),
106                Value::OptionJson(v) => q.bind(v.map(|j| j.to_string())),
107                Value::OptionJsonb(v) => q.bind(v.map(|j| j.to_string())),
108            };
109        }
110        q
111    }};
112}
113
114// ═══════════════════════════════════════════════════════════════
115// impl Executor for Db
116// ═══════════════════════════════════════════════════════════════
117
118impl Executor for Db {
119    #[allow(unreachable_patterns)]
120    async fn execute_raw(&self, sql: &str, params: Vec<Value>) -> Result<u64, FlozError> {
121        let sql = self.adjust_sql(sql);
122        match &self.inner {
123            #[cfg(feature = "postgres")]
124            DbInner::Postgres(pool) => {
125                let query = bind_params!(sqlx::query(&sql), params);
126                let result = query.execute(pool).await?;
127                Ok(result.rows_affected())
128            }
129            #[cfg(feature = "sqlite")]
130            DbInner::Sqlite(pool) => {
131                let query = bind_params!(sqlx::query(&sql), params);
132                let result = query.execute(pool).await?;
133                Ok(result.rows_affected())
134            }
135            _ => unreachable!("no database backend enabled"),
136        }
137    }
138
139    #[allow(unreachable_patterns)]
140    async fn fetch_all<T>(&self, sql: &str, params: Vec<Value>) -> Result<Vec<T>, FlozError>
141    where
142        T: Send + Unpin + FlozRowBound,
143    {
144        let sql = self.adjust_sql(sql);
145        match &self.inner {
146            #[cfg(feature = "postgres")]
147            DbInner::Postgres(pool) => {
148                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
149                Ok(query.fetch_all(pool).await?)
150            }
151            #[cfg(feature = "sqlite")]
152            DbInner::Sqlite(pool) => {
153                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
154                Ok(query.fetch_all(pool).await?)
155            }
156            _ => unreachable!("no database backend enabled"),
157        }
158    }
159
160    #[allow(unreachable_patterns)]
161    async fn fetch_one<T>(&self, sql: &str, params: Vec<Value>) -> Result<T, FlozError>
162    where
163        T: Send + Unpin + FlozRowBound,
164    {
165        let sql = self.adjust_sql(sql);
166        match &self.inner {
167            #[cfg(feature = "postgres")]
168            DbInner::Postgres(pool) => {
169                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
170                query
171                    .fetch_optional(pool)
172                    .await?
173                    .ok_or(FlozError::NotFound)
174            }
175            #[cfg(feature = "sqlite")]
176            DbInner::Sqlite(pool) => {
177                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
178                query
179                    .fetch_optional(pool)
180                    .await?
181                    .ok_or(FlozError::NotFound)
182            }
183            _ => unreachable!("no database backend enabled"),
184        }
185    }
186
187    #[allow(unreachable_patterns)]
188    async fn fetch_optional<T>(
189        &self,
190        sql: &str,
191        params: Vec<Value>,
192    ) -> Result<Option<T>, FlozError>
193    where
194        T: Send + Unpin + FlozRowBound,
195    {
196        let sql = self.adjust_sql(sql);
197        match &self.inner {
198            #[cfg(feature = "postgres")]
199            DbInner::Postgres(pool) => {
200                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
201                Ok(query.fetch_optional(pool).await?)
202            }
203            #[cfg(feature = "sqlite")]
204            DbInner::Sqlite(pool) => {
205                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
206                Ok(query.fetch_optional(pool).await?)
207            }
208            _ => unreachable!("no database backend enabled"),
209        }
210    }
211}
212
213// ═══════════════════════════════════════════════════════════════
214// impl Executor for Tx
215// ═══════════════════════════════════════════════════════════════
216
217impl Executor for Tx {
218    #[allow(unreachable_patterns)]
219    async fn execute_raw(&self, sql: &str, params: Vec<Value>) -> Result<u64, FlozError> {
220        let sql = self.adjust_sql(sql);
221        match &self.inner {
222            #[cfg(feature = "postgres")]
223            TxInner::Postgres(conn) => {
224                let mut c = conn.lock().await;
225                let query = bind_params!(sqlx::query(&sql), params);
226                let result = query.execute(&mut **c).await?;
227                Ok(result.rows_affected())
228            }
229            #[cfg(feature = "sqlite")]
230            TxInner::Sqlite(conn) => {
231                let mut c = conn.lock().await;
232                let query = bind_params!(sqlx::query(&sql), params);
233                let result = query.execute(&mut **c).await?;
234                Ok(result.rows_affected())
235            }
236            _ => unreachable!("no database backend enabled"),
237        }
238    }
239
240    #[allow(unreachable_patterns)]
241    async fn fetch_all<T>(&self, sql: &str, params: Vec<Value>) -> Result<Vec<T>, FlozError>
242    where
243        T: Send + Unpin + FlozRowBound,
244    {
245        let sql = self.adjust_sql(sql);
246        match &self.inner {
247            #[cfg(feature = "postgres")]
248            TxInner::Postgres(conn) => {
249                let mut c = conn.lock().await;
250                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
251                Ok(query.fetch_all(&mut **c).await?)
252            }
253            #[cfg(feature = "sqlite")]
254            TxInner::Sqlite(conn) => {
255                let mut c = conn.lock().await;
256                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
257                Ok(query.fetch_all(&mut **c).await?)
258            }
259            _ => unreachable!("no database backend enabled"),
260        }
261    }
262
263    #[allow(unreachable_patterns)]
264    async fn fetch_one<T>(&self, sql: &str, params: Vec<Value>) -> Result<T, FlozError>
265    where
266        T: Send + Unpin + FlozRowBound,
267    {
268        let sql = self.adjust_sql(sql);
269        match &self.inner {
270            #[cfg(feature = "postgres")]
271            TxInner::Postgres(conn) => {
272                let mut c = conn.lock().await;
273                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
274                query
275                    .fetch_optional(&mut **c)
276                    .await?
277                    .ok_or(FlozError::NotFound)
278            }
279            #[cfg(feature = "sqlite")]
280            TxInner::Sqlite(conn) => {
281                let mut c = conn.lock().await;
282                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
283                query
284                    .fetch_optional(&mut **c)
285                    .await?
286                    .ok_or(FlozError::NotFound)
287            }
288            _ => unreachable!("no database backend enabled"),
289        }
290    }
291
292    #[allow(unreachable_patterns)]
293    async fn fetch_optional<T>(
294        &self,
295        sql: &str,
296        params: Vec<Value>,
297    ) -> Result<Option<T>, FlozError>
298    where
299        T: Send + Unpin + FlozRowBound,
300    {
301        let sql = self.adjust_sql(sql);
302        match &self.inner {
303            #[cfg(feature = "postgres")]
304            TxInner::Postgres(conn) => {
305                let mut c = conn.lock().await;
306                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
307                Ok(query.fetch_optional(&mut **c).await?)
308            }
309            #[cfg(feature = "sqlite")]
310            TxInner::Sqlite(conn) => {
311                let mut c = conn.lock().await;
312                let query = bind_params!(sqlx::query_as::<_, T>(&sql), params);
313                Ok(query.fetch_optional(&mut **c).await?)
314            }
315            _ => unreachable!("no database backend enabled"),
316        }
317    }
318}
319
320// ═══════════════════════════════════════════════════════════════
321// Tests
322// ═══════════════════════════════════════════════════════════════
323
324#[cfg(test)]
325mod tests {
326    use super::*;
327
328    #[test]
329    fn db_is_send_sync() {
330        fn _assert_send<T: Send>() {}
331        fn _assert_sync<T: Sync>() {}
332        _assert_send::<Db>();
333        _assert_sync::<Db>();
334    }
335
336    #[test]
337    fn tx_is_debug() {
338        fn _assert_debug<T: std::fmt::Debug>() {}
339        _assert_debug::<Tx>();
340    }
341
342    // ── SQLite integration tests ──
343
344    #[cfg(feature = "sqlite")]
345    mod sqlite_tests {
346        use super::*;
347
348        #[tokio::test]
349        async fn connect_memory() {
350            let db = Db::connect("sqlite::memory:").await.unwrap();
351            let affected = db.execute_raw("SELECT 1", vec![]).await.unwrap();
352            assert_eq!(affected, 0);
353        }
354
355        #[tokio::test]
356        async fn create_insert_fetch() {
357            let db = Db::connect("sqlite::memory:").await.unwrap();
358
359            db.execute_raw(
360                "CREATE TABLE test_users (id INTEGER PRIMARY KEY, name TEXT NOT NULL)",
361                vec![],
362            )
363            .await
364            .unwrap();
365
366            let affected = db
367                .execute_raw(
368                    "INSERT INTO test_users (id, name) VALUES ($1, $2)",
369                    vec![Value::Int(1), Value::String("Alice".into())],
370                )
371                .await
372                .unwrap();
373            assert_eq!(affected, 1);
374
375            #[derive(Debug, sqlx::FromRow)]
376            struct User {
377                id: i32,
378                name: String,
379            }
380
381            let users: Vec<User> = db
382                .fetch_all("SELECT id, name FROM test_users ORDER BY id", vec![])
383                .await
384                .unwrap();
385            assert_eq!(users.len(), 1);
386            assert_eq!(users[0].name, "Alice");
387        }
388
389        #[tokio::test]
390        async fn fetch_one_not_found() {
391            let db = Db::connect("sqlite::memory:").await.unwrap();
392            db.execute_raw("CREATE TABLE test_nf (id INTEGER)", vec![])
393                .await
394                .unwrap();
395
396            #[derive(Debug, sqlx::FromRow)]
397            struct Row {
398                id: i32,
399            }
400
401            let result: Result<Row, _> = db
402                .fetch_one(
403                    "SELECT id FROM test_nf WHERE id = $1",
404                    vec![Value::Int(999)],
405                )
406                .await;
407            assert!(matches!(result, Err(FlozError::NotFound)));
408        }
409
410        #[tokio::test]
411        async fn transaction_commit() {
412            let db = Db::connect("sqlite::memory:").await.unwrap();
413            db.execute_raw("CREATE TABLE test_tx (id INTEGER)", vec![])
414                .await
415                .unwrap();
416
417            let tx = db.begin().await.unwrap();
418            tx.execute_raw(
419                "INSERT INTO test_tx VALUES ($1)",
420                vec![Value::Int(1)],
421            )
422            .await
423            .unwrap();
424            tx.commit().await.unwrap();
425
426            #[derive(Debug, sqlx::FromRow)]
427            struct Row {
428                id: i32,
429            }
430
431            let rows: Vec<Row> = db
432                .fetch_all("SELECT id FROM test_tx", vec![])
433                .await
434                .unwrap();
435            assert_eq!(rows.len(), 1);
436        }
437
438        #[tokio::test]
439        async fn transaction_rollback() {
440            let db = Db::connect("sqlite::memory:").await.unwrap();
441            db.execute_raw("CREATE TABLE test_rb (id INTEGER)", vec![])
442                .await
443                .unwrap();
444
445            let tx = db.begin().await.unwrap();
446            tx.execute_raw(
447                "INSERT INTO test_rb VALUES ($1)",
448                vec![Value::Int(1)],
449            )
450            .await
451            .unwrap();
452            tx.rollback().await.unwrap();
453
454            #[derive(Debug, sqlx::FromRow)]
455            struct Row {
456                id: i32,
457            }
458
459            let rows: Vec<Row> = db
460                .fetch_all("SELECT id FROM test_rb", vec![])
461                .await
462                .unwrap();
463            assert_eq!(rows.len(), 0);
464        }
465    }
466
467    // ── PostgreSQL integration tests ──
468
469    #[cfg(feature = "postgres")]
470    mod pg_tests {
471        use super::*;
472
473        fn test_url() -> Option<String> {
474            std::env::var("DATABASE_URL").ok()
475        }
476
477        #[tokio::test]
478        async fn connect_to_postgres() {
479            let Some(url) = test_url() else { return };
480            let db = Db::connect(&url).await.unwrap();
481            let affected = db.execute_raw("SELECT 1", vec![]).await.unwrap();
482            assert_eq!(affected, 0);
483        }
484
485        #[tokio::test]
486        async fn execute_with_params() {
487            let Some(url) = test_url() else { return };
488            let db = Db::connect(&url).await.unwrap();
489
490            db.execute_raw(
491                "CREATE TEMP TABLE floz_test_params (id INT, name TEXT)",
492                vec![],
493            )
494            .await
495            .unwrap();
496
497            let affected = db
498                .execute_raw(
499                    "INSERT INTO floz_test_params (id, name) VALUES ($1, $2)",
500                    vec![Value::Int(1), Value::String("Alice".into())],
501                )
502                .await
503                .unwrap();
504            assert_eq!(affected, 1);
505        }
506
507        #[tokio::test]
508        async fn fetch_all_rows() {
509            let Some(url) = test_url() else { return };
510            let db = Db::connect(&url).await.unwrap();
511
512            db.execute_raw(
513                "CREATE TEMP TABLE floz_test_fetch (id INT, name TEXT)",
514                vec![],
515            )
516            .await
517            .unwrap();
518            db.execute_raw(
519                "INSERT INTO floz_test_fetch VALUES (1, 'Alice'), (2, 'Bob')",
520                vec![],
521            )
522            .await
523            .unwrap();
524
525            #[derive(Debug, sqlx::FromRow)]
526            struct Row {
527                id: i32,
528                name: String,
529            }
530
531            let rows: Vec<Row> = db
532                .fetch_all(
533                    "SELECT id, name FROM floz_test_fetch ORDER BY id",
534                    vec![],
535                )
536                .await
537                .unwrap();
538            assert_eq!(rows.len(), 2);
539            assert_eq!(rows[0].name, "Alice");
540            assert_eq!(rows[1].name, "Bob");
541        }
542
543        #[tokio::test]
544        async fn transaction_commit() {
545            let Some(url) = test_url() else { return };
546            let db = Db::connect(&url).await.unwrap();
547
548            db.execute_raw("CREATE TEMP TABLE floz_test_tx (id INT)", vec![])
549                .await
550                .unwrap();
551
552            let tx = db.begin().await.unwrap();
553            tx.execute_raw(
554                "INSERT INTO floz_test_tx VALUES ($1)",
555                vec![Value::Int(1)],
556            )
557            .await
558            .unwrap();
559            tx.commit().await.unwrap();
560
561            #[derive(Debug, sqlx::FromRow)]
562            struct Row {
563                id: i32,
564            }
565
566            let rows: Vec<Row> = db
567                .fetch_all("SELECT id FROM floz_test_tx", vec![])
568                .await
569                .unwrap();
570            assert_eq!(rows.len(), 1);
571        }
572    }
573}
floz_orm/db/executor.rs

floz_orm/db/
executor.rs
