laterite_core/
query.rs

1//! Portable parameterised queries over `sqlx::Any`.
2//!
3//! `sea-query` builds the SQL and an ordered list of values for the running
4//! backend; [`bind_values`] binds those values to a `sqlx::Any` query. Only
5//! portable value kinds are supported (bool, integers, floats, text, bytes), so
6//! backend-specific types (timestamps, JSON) are represented as text or integers
7//! at the query boundary and converted in Rust.
8
9use sqlx::any::{AnyArguments, AnyRow};
10use sqlx::{Any, Encode, Row, Type};
11
12use crate::migration::DbBackend;
13use crate::Db;
14
15/// Row helpers for portably stored types.
16pub trait AnyRowExt {
17    /// Reads a boolean stored as a 0/1 integer (the portable representation, see
18    /// [`bind_values`]). The counterpart to binding a `bool`.
19    fn get_bool(&self, column: &str) -> Result<bool, sqlx::Error>;
20
21    /// Reads a text column as a `String`, portably. MySQL reports `TEXT` columns
22    /// as `BLOB` through `sqlx::Any`, so a plain `String` decode fails there; this
23    /// falls back to reading the bytes and decoding UTF-8. Use for every text read
24    /// so the same code works on every backend.
25    fn get_text(&self, column: &str) -> Result<String, sqlx::Error>;
26
27    /// The nullable counterpart to [`get_text`], for a `text` column that may be
28    /// `NULL`.
29    fn get_text_opt(&self, column: &str) -> Result<Option<String>, sqlx::Error>;
30
31    /// Reads an integer column as `i64`, portably. `sqlx::Any` reports SQLite
32    /// integers as `i64` but Postgres `INTEGER` as `i32`, so a fixed width fails
33    /// on one backend; this tries `i64` then falls back to `i32`. Use for any
34    /// integer column whatever its declared width, so a column need not be widened
35    /// to `bigint` just to be read back.
36    fn get_int(&self, column: &str) -> Result<i64, sqlx::Error>;
37
38    /// The nullable counterpart to [`get_int`].
39    fn get_int_opt(&self, column: &str) -> Result<Option<i64>, sqlx::Error>;
40}
41
42impl AnyRowExt for AnyRow {
43    fn get_bool(&self, column: &str) -> Result<bool, sqlx::Error> {
44        Ok(self.try_get::<i32, _>(column)? != 0)
45    }
46
47    fn get_text(&self, column: &str) -> Result<String, sqlx::Error> {
48        match self.try_get::<String, _>(column) {
49            Ok(s) => Ok(s),
50            Err(_) => decode_utf8(column, self.try_get::<Vec<u8>, _>(column)?),
51        }
52    }
53
54    fn get_text_opt(&self, column: &str) -> Result<Option<String>, sqlx::Error> {
55        match self.try_get::<Option<String>, _>(column) {
56            Ok(s) => Ok(s),
57            Err(_) => self
58                .try_get::<Option<Vec<u8>>, _>(column)?
59                .map(|b| decode_utf8(column, b))
60                .transpose(),
61        }
62    }
63
64    fn get_int(&self, column: &str) -> Result<i64, sqlx::Error> {
65        match self.try_get::<i64, _>(column) {
66            Ok(v) => Ok(v),
67            Err(_) => Ok(i64::from(self.try_get::<i32, _>(column)?)),
68        }
69    }
70
71    fn get_int_opt(&self, column: &str) -> Result<Option<i64>, sqlx::Error> {
72        match self.try_get::<Option<i64>, _>(column) {
73            Ok(v) => Ok(v),
74            Err(_) => Ok(self.try_get::<Option<i32>, _>(column)?.map(i64::from)),
75        }
76    }
77}
78
79/// Decodes bytes read from a text-typed column (the MySQL `BLOB` fallback path)
80/// as UTF-8, surfacing a bad encoding as a column-decode error.
81fn decode_utf8(column: &str, bytes: Vec<u8>) -> Result<String, sqlx::Error> {
82    String::from_utf8(bytes).map_err(|e| sqlx::Error::ColumnDecode {
83        index: column.to_string(),
84        source: Box::new(e),
85    })
86}
87
88/// A portable "insert, ignore duplicates" conflict clause for `keys` (the unique
89/// or primary-key columns of the conflict). Use in place of
90/// `OnConflict::columns(keys).do_nothing()`: sea-query renders MySQL's
91/// `DO NOTHING` as invalid SQL (`ON DUPLICATE KEY IGNORE`), so this expresses the
92/// same intent as a no-op update of the first key column, which is valid on
93/// Postgres, MySQL, and SQLite alike.
94pub fn on_conflict_ignore<C>(keys: impl IntoIterator<Item = C>) -> sea_query::OnConflict
95where
96    C: sea_query::IntoIden,
97{
98    use sea_query::IntoIden;
99    let keys: Vec<sea_query::DynIden> = keys.into_iter().map(IntoIden::into_iden).collect();
100    let first = keys[0].clone();
101    sea_query::OnConflict::columns(keys)
102        .update_column(first)
103        .to_owned()
104}
105
106/// The SQL type to cast a column to when you need its value back as a string on
107/// any backend. Descriptor-driven screens cast every selected column to a string
108/// so a value of any type reads back uniformly through `sqlx::Any`. MySQL rejects
109/// `CAST(x AS text)` (it casts to `char`), while Postgres and SQLite use `text`.
110/// Use as `Expr::col(c).cast_as(sea_query::Alias::new(text_cast(backend)))`.
111pub fn text_cast(backend: DbBackend) -> &'static str {
112    match backend {
113        DbBackend::Mysql => "char",
114        DbBackend::Postgres | DbBackend::Sqlite => "text",
115    }
116}
117
118/// Renders a `sea-query` statement to `(sql, values)` for `backend`.
119///
120/// The statement is taken by value and dropped here, so no `sea-query` builder
121/// (which holds non-`Send` reference-counted identifiers) survives into the
122/// caller's `.await`. That keeps request handlers' futures `Send`, as the async
123/// runtime requires. Construct the statement, hand it here, then bind and run
124/// the returned owned `(sql, values)`.
125pub fn build<S>(backend: DbBackend, stmt: S) -> (String, sea_query::Values)
126where
127    S: sea_query::QueryStatementWriter,
128{
129    match backend {
130        DbBackend::Postgres => stmt.build(sea_query::PostgresQueryBuilder),
131        DbBackend::Mysql => stmt.build(sea_query::MysqlQueryBuilder),
132        DbBackend::Sqlite => stmt.build(sea_query::SqliteQueryBuilder),
133    }
134}
135
136/// Runs an insert and returns the generated auto-increment id, portably.
137///
138/// The id column is `bigint auto_increment` on every backend, but reading the
139/// new id back is not uniform: Postgres exposes no last-insert-id, so its
140/// statement is given a `RETURNING` clause and the id is read from the returned
141/// row; MySQL and SQLite report it through the driver after a plain execute.
142/// `id` names the auto-increment column (used only for the Postgres `RETURNING`).
143//
144// This is a synchronous function returning a future, not an `async fn`: an
145// `async fn` captures all its parameters into the future from the moment it is
146// created, so taking the non-`Send` `sea-query` builder by value would make the
147// future non-`Send` regardless of when the body drops it. Rendering the builder
148// to owned `Send` data first, then capturing only that into an `async move`
149// block, keeps the returned future `Send` as request handlers require.
150pub fn insert_returning_id<I>(
151    db: &Db,
152    stmt: sea_query::InsertStatement,
153    id: I,
154) -> impl std::future::Future<Output = Result<i64, sqlx::Error>> + Send + '_
155where
156    I: sea_query::IntoIden + 'static,
157{
158    let (sql, values, returning) = render_insert(db.backend, stmt, id);
159    async move {
160        if returning {
161            bind_values(sqlx::query(&sql), values)
162                .fetch_one(&db.pool)
163                .await?
164                .try_get::<i64, _>(0)
165        } else {
166            bind_values(sqlx::query(&sql), values)
167                .execute(&db.pool)
168                .await?
169                .last_insert_id()
170                .ok_or(sqlx::Error::RowNotFound)
171        }
172    }
173}
174
175/// Like [`insert_returning_id`] but on an explicit connection (a transaction), so
176/// a multi-statement write stays atomic. Sync-renders then captures only owned
177/// `Send` data plus the connection into the future, for the same reason (see
178/// [`insert_returning_id`]).
179pub fn insert_returning_id_on<'c, I>(
180    backend: DbBackend,
181    conn: &'c mut sqlx::AnyConnection,
182    stmt: sea_query::InsertStatement,
183    id: I,
184) -> impl std::future::Future<Output = Result<i64, sqlx::Error>> + Send + 'c
185where
186    I: sea_query::IntoIden + 'static,
187{
188    let (sql, values, returning) = render_insert(backend, stmt, id);
189    async move {
190        if returning {
191            bind_values(sqlx::query(&sql), values)
192                .fetch_one(&mut *conn)
193                .await?
194                .try_get::<i64, _>(0)
195        } else {
196            bind_values(sqlx::query(&sql), values)
197                .execute(&mut *conn)
198                .await?
199                .last_insert_id()
200                .ok_or(sqlx::Error::RowNotFound)
201        }
202    }
203}
204
205/// Renders an insert to `(sql, values, use_returning)`. Postgres and SQLite get a
206/// `RETURNING` clause on `id` (read back with `fetch_one`); MySQL has no portable
207/// `RETURNING`, so it reports the id through the driver's last-insert-id instead.
208fn render_insert<I>(
209    backend: DbBackend,
210    mut stmt: sea_query::InsertStatement,
211    id: I,
212) -> (String, sea_query::Values, bool)
213where
214    I: sea_query::IntoIden + 'static,
215{
216    let returning = matches!(backend, DbBackend::Postgres | DbBackend::Sqlite);
217    if returning {
218        stmt.returning_col(id);
219    }
220    let (sql, values) = build(backend, stmt);
221    (sql, values, returning)
222}
223
224type AnyQuery<'q> = sqlx::query::Query<'q, Any, AnyArguments<'q>>;
225type AnyQueryAs<'q, O> = sqlx::query::QueryAs<'q, Any, O, AnyArguments<'q>>;
226
227fn bind_one<'q, T>(query: AnyQuery<'q>, value: T) -> AnyQuery<'q>
228where
229    T: 'q + Send + Encode<'q, Any> + Type<Any>,
230{
231    query.bind(value)
232}
233
234/// Binds `sea-query` values onto a `sqlx::Any` query, in order. Only portable
235/// value kinds occur here (the framework converts time/JSON to text before
236/// building), so an unsupported kind is a framework bug and panics.
237pub fn bind_values(mut query: AnyQuery<'_>, values: sea_query::Values) -> AnyQuery<'_> {
238    use sea_query::Value;
239    for value in values.0 {
240        query = match value {
241            // Booleans are stored as 0/1 integers everywhere (a SQLite `boolean`
242            // column is not decodable through `sqlx::Any`), so a `bool` value
243            // binds as an integer. Callers write `bool`; storage stays portable.
244            Value::Bool(v) => bind_one(query, v.map(i32::from)),
245            Value::TinyInt(v) => bind_one(query, v.map(i32::from)),
246            Value::SmallInt(v) => bind_one(query, v),
247            Value::Int(v) => bind_one(query, v),
248            Value::BigInt(v) => bind_one(query, v),
249            // `sqlx::Any` has no unsigned column types, so unsigned integers
250            // (sea-query emits these for LIMIT/OFFSET) bind as the next wider
251            // signed integer. The framework never stores unsigned values.
252            Value::TinyUnsigned(v) => bind_one(query, v.map(i32::from)),
253            Value::SmallUnsigned(v) => bind_one(query, v.map(i32::from)),
254            Value::Unsigned(v) => bind_one(query, v.map(i64::from)),
255            Value::BigUnsigned(v) => bind_one(query, v.map(|n| n as i64)),
256            Value::Float(v) => bind_one(query, v),
257            Value::Double(v) => bind_one(query, v),
258            Value::String(v) => bind_one(query, v.map(|b| *b)),
259            Value::Char(v) => bind_one(query, v.map(|c| c.to_string())),
260            Value::Bytes(v) => bind_one(query, v.map(|b| *b)),
261            // Defensive: unreachable under the framework's sea-query features, but
262            // guards against a richer `Value` if a backend feature is unified in.
263            #[allow(unreachable_patterns)]
264            other => panic!("unsupported portable bind value: {other:?}"),
265        };
266    }
267    query
268}
269
270/// Same as [`bind_values`], for a `query_as` mapping rows into `O`.
271pub fn bind_values_as<O>(
272    mut query: AnyQueryAs<'_, O>,
273    values: sea_query::Values,
274) -> AnyQueryAs<'_, O> {
275    use sea_query::Value;
276    for value in values.0 {
277        query = match value {
278            Value::Bool(v) => query.bind(v.map(i32::from)),
279            Value::TinyInt(v) => query.bind(v.map(i32::from)),
280            Value::SmallInt(v) => query.bind(v),
281            Value::Int(v) => query.bind(v),
282            Value::BigInt(v) => query.bind(v),
283            Value::TinyUnsigned(v) => query.bind(v.map(i32::from)),
284            Value::SmallUnsigned(v) => query.bind(v.map(i32::from)),
285            Value::Unsigned(v) => query.bind(v.map(i64::from)),
286            Value::BigUnsigned(v) => query.bind(v.map(|n| n as i64)),
287            Value::Float(v) => query.bind(v),
288            Value::Double(v) => query.bind(v),
289            Value::String(v) => query.bind(v.map(|b| *b)),
290            Value::Char(v) => query.bind(v.map(|c| c.to_string())),
291            Value::Bytes(v) => query.bind(v.map(|b| *b)),
292            #[allow(unreachable_patterns)]
293            other => panic!("unsupported portable bind value: {other:?}"),
294        };
295    }
296    query
297}
298
299#[cfg(test)]
300mod tests {
301    use super::*;
302    use sea_query::{Alias, Expr, Iden, Query};
303    use sqlx::AnyPool;
304
305    #[derive(Iden)]
306    enum Widget {
307        Table,
308        Id,
309        Label,
310        Qty,
311    }
312
313    async fn pool() -> AnyPool {
314        sqlx::any::install_default_drivers();
315        let pool = sqlx::any::AnyPoolOptions::new()
316            .max_connections(1)
317            .connect("sqlite::memory:")
318            .await
319            .unwrap();
320        sqlx::raw_sql(
321            "create table widget (id text primary key, label text not null, qty integer not null)",
322        )
323        .execute(&pool)
324        .await
325        .unwrap();
326        pool
327    }
328
329    #[tokio::test]
330    async fn binds_parameters_on_insert_and_select() {
331        let pool = pool().await;
332        let backend = DbBackend::Sqlite;
333
334        let insert = Query::insert()
335            .into_table(Widget::Table)
336            .columns([Widget::Id, Widget::Label, Widget::Qty])
337            .values_panic(["w-1".into(), "Sprocket".into(), 7.into()])
338            .to_owned();
339        let (sql, values) = build(backend, insert);
340        bind_values(sqlx::query(&sql), values)
341            .execute(&pool)
342            .await
343            .unwrap();
344
345        // Select the label back by a bound id parameter.
346        let select = Query::select()
347            .column(Widget::Label)
348            .from(Widget::Table)
349            .and_where(Expr::col(Widget::Id).eq("w-1"))
350            .to_owned();
351        let (sql, values) = build(backend, select);
352        let label: String = bind_values_as(sqlx::query_as::<_, (String,)>(&sql), values)
353            .fetch_one(&pool)
354            .await
355            .unwrap()
356            .0;
357        assert_eq!(label, "Sprocket");
358
359        // A cast-free count via a bound filter.
360        let count_stmt = Query::select()
361            .expr(Expr::col(Widget::Id).count())
362            .from(Widget::Table)
363            .and_where(Expr::col(Alias::new("qty")).eq(7))
364            .to_owned();
365        let (sql, values) = build(backend, count_stmt);
366        let count: i64 = bind_values_as(sqlx::query_as::<_, (i64,)>(&sql), values)
367            .fetch_one(&pool)
368            .await
369            .unwrap()
370            .0;
371        assert_eq!(count, 1);
372    }
373
374    // Runs on the full backend matrix: the id read-back differs per backend
375    // (Postgres RETURNING vs MySQL last-insert-id), so it uses the portable test
376    // harness + schema builder rather than raw, SQLite-only DDL.
377    #[cfg(feature = "testing")]
378    #[tokio::test]
379    async fn insert_returning_id_on_runs_inside_a_transaction() {
380        use crate::strata::{
381            async_trait, ColumnDef, CoreResult, Migration, MigrationSet, Schema, Table,
382        };
383        use crate::testing::connect_test;
384
385        struct CreateThing;
386        #[async_trait(?Send)]
387        impl Migration for CreateThing {
388            fn name(&self) -> &str {
389                "0001_create_thing"
390            }
391            async fn up(&self, s: &mut Schema<'_>) -> CoreResult<()> {
392                s.exec(
393                    Table::create()
394                        .table(Alias::new("thing"))
395                        .if_not_exists()
396                        .col(
397                            ColumnDef::new(Alias::new("id"))
398                                .big_integer()
399                                .not_null()
400                                .auto_increment()
401                                .primary_key(),
402                        )
403                        .col(ColumnDef::new(Alias::new("name")).text().not_null())
404                        .to_owned(),
405                )
406                .await
407            }
408        }
409
410        let (db, _guard) =
411            connect_test(&[MigrationSet::new("test.thing", vec![Box::new(CreateThing)])]).await;
412
413        // Insert inside a transaction, reading the new id back through the tx.
414        let mut tx = db.pool.begin().await.unwrap();
415        let insert = Query::insert()
416            .into_table(Alias::new("thing"))
417            .columns([Alias::new("name")])
418            .values_panic(["gadget".into()])
419            .to_owned();
420        let id = insert_returning_id_on(db.backend, &mut tx, insert, Alias::new("id"))
421            .await
422            .unwrap();
423        tx.commit().await.unwrap();
424        assert!(id >= 1);
425
426        // The committed row is readable by the returned id.
427        let select = Query::select()
428            .column(Alias::new("name"))
429            .from(Alias::new("thing"))
430            .and_where(Expr::col(Alias::new("id")).eq(id))
431            .to_owned();
432        let (sql, values) = build(db.backend, select);
433        let name: String = bind_values_as(sqlx::query_as::<_, (String,)>(&sql), values)
434            .fetch_one(&db.pool)
435            .await
436            .unwrap()
437            .0;
438        assert_eq!(name, "gadget");
439    }
440}
laterite_core/query.rs

laterite_core/
query.rs
