qbrs_sqlx/
lib.rs

1//! Execution integration between qbrs's query builder and a real Postgres
2//! via `sqlx`. This crate owns only the value-binding and row-decoding glue;
3//! query building, SQL rendering, and every compile-time guarantee live in
4//! `qbrs-core`, which stays independent of any async runtime or driver.
5
6/// Re-exported so a caller can name what `LoadExt::stream` returns and
7/// consume it (`StreamExt::next` is how a stream is read) without taking
8/// a `futures` dependency of their own.
9pub use futures_util::{Stream, StreamExt};
10use qbrs_core::delete::Delete;
11use qbrs_core::dialect::Postgres;
12use qbrs_core::expr::Value;
13use qbrs_core::insert::Insert;
14use qbrs_core::row::{Row, RowCons, RowNil};
15use qbrs_core::select::{DynSelect, Prepared, PreparedParams, Select, Selection, SetOp, Total};
16use qbrs_core::statement::{Returning, Statement, WrittenTable};
17use qbrs_core::update::Update;
18use sqlx::Row as _;
19use sqlx::postgres::PgRow;
20
21/// Errors from executing a qbrs query against Postgres via `sqlx`. An enum
22/// rather than a bare `sqlx::Error` so a qbrs-level misuse is distinguishable
23/// from a driver/database error without string-matching a message.
24#[derive(Debug, thiserror::Error)]
25pub enum Error {
26    /// A real error from Postgres or the `sqlx` driver: a failed
27    /// connection, constraint violation, decode failure, etc.
28    #[error(transparent)]
29    Sqlx(#[from] sqlx::Error),
30
31    /// A `prepare!{}` placeholder reached execution unresolved: either
32    /// `Prepared::resolve()` found no matching field in `Params`, or the
33    /// query was executed directly instead of through `.prepare()`.
34    #[error(transparent)]
35    UnresolvedPlaceholder(#[from] qbrs_core::select::UnresolvedPlaceholder),
36
37    /// An `*Update` describing no assignment, or an insert of no rows.
38    /// Caught where the request-shaped data is read (`Assignments::from_row`,
39    /// `.values_all`), never at a statement. Here so a handler returning
40    /// this crate's `Result` can `?` on that as readily as on a query.
41    #[error(transparent)]
42    NothingToSet(#[from] qbrs_core::update::NothingToSet),
43
44    #[error(transparent)]
45    NothingToInsert(#[from] qbrs_core::insert::NothingToInsert),
46
47    /// A column type is enabled on `qbrs` but not on `qbrs-sqlx`, so the
48    /// value renders and has nothing to bind it. The two crates carry the
49    /// same feature names for exactly this reason: turn it on in both.
50    #[error("`{0}` values need the matching feature on `qbrs-sqlx` too")]
51    FeatureNotEnabled(&'static str),
52}
53
54/// This crate's `Result`: the same shape as `sqlx::Result`, with
55/// `qbrs_sqlx::Error` as the fixed error type.
56pub type Result<T> = std::result::Result<T, Error>;
57
58/// Every extension trait that puts a terminal method on a builder, plus the
59/// error type a caller's own signatures have to name and the `DecodeRow`
60/// bound a generic helper over `RowQuery` has to spell. `Result` is
61/// deliberately absent: a glob-imported alias of that name shadows
62/// `std::result::Result` in every module that follows, and a service layer
63/// has its own error type in most of them. Write `qbrs_sqlx::Result<T>`
64/// where the alias is wanted. Which trait applies depends on the builder,
65/// so importing them one at a time is bookkeeping with no decision in it.
66/// `count` in particular resolves against `Iterator::count` with a
67/// confusing message until `CountExt` is in scope.
68pub mod prelude {
69    pub use crate::Error;
70    pub use crate::{
71        CountExt, CountQuery, DecodeRow, ExecuteExt, LoadExt, PreparedCountExt, PreparedExt,
72        PreparedQuery, PreparedTotal, RowQuery, Stream, StreamExt, WriteStatement,
73    };
74}
75
76/// Binds a `Value` to a Postgres query parameter. `Value`'s typed `NullX`
77/// variants carry the parameter type a NULL bind still has to declare.
78///
79/// Fallible only for `Value::Placeholder`: an unresolved named placeholder
80/// is a misuse no compile-time check here can catch, so it surfaces as
81/// `Error::UnresolvedPlaceholder` rather than a wrong bind or a panic.
82fn bind_value<'q>(
83    query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
84    v: Value,
85) -> Result<sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>> {
86    Ok(match v {
87        Value::I32(x) => query.bind(x),
88        Value::I64(x) => query.bind(x),
89        Value::F64(x) => query.bind(x),
90        Value::Text(x) => query.bind(x),
91        Value::Bool(x) => query.bind(x),
92        Value::Bytes(x) => query.bind(x),
93        Value::NullI32 => query.bind(None::<i32>),
94        Value::NullI64 => query.bind(None::<i64>),
95        Value::NullF64 => query.bind(None::<f64>),
96        Value::NullText => query.bind(None::<String>),
97        Value::NullBool => query.bind(None::<bool>),
98        Value::NullBytes => query.bind(None::<Vec<u8>>),
99        Value::TextArray(x) => query.bind(x),
100        Value::NullTextArray => query.bind(None::<Vec<String>>),
101        Value::IntegerArray(x) => query.bind(x),
102        Value::NullIntegerArray => query.bind(None::<Vec<i32>>),
103        Value::BigIntArray(x) => query.bind(x),
104        Value::NullBigIntArray => query.bind(None::<Vec<i64>>),
105        #[cfg(feature = "uuid")]
106        Value::UuidArray(x) => query.bind(x),
107        #[cfg(feature = "uuid")]
108        Value::NullUuidArray => query.bind(None::<Vec<uuid::Uuid>>),
109        #[cfg(feature = "json")]
110        Value::Json(x) => query.bind(x),
111        #[cfg(feature = "json")]
112        Value::NullJson => query.bind(None::<serde_json::Value>),
113        #[cfg(feature = "chrono")]
114        Value::Timestamptz(x) => query.bind(x),
115        #[cfg(feature = "chrono")]
116        Value::NullTimestamptz => query.bind(None::<chrono::DateTime<chrono::Utc>>),
117        #[cfg(feature = "chrono")]
118        Value::Date(x) => query.bind(x),
119        #[cfg(feature = "chrono")]
120        Value::NullDate => query.bind(None::<chrono::NaiveDate>),
121        #[cfg(feature = "uuid")]
122        Value::Uuid(x) => query.bind(x),
123        #[cfg(feature = "uuid")]
124        Value::NullUuid => query.bind(None::<uuid::Uuid>),
125        #[cfg(feature = "decimal")]
126        Value::Numeric(x) => query.bind(x),
127        #[cfg(feature = "decimal")]
128        Value::NullNumeric => query.bind(None::<rust_decimal::Decimal>),
129        Value::Placeholder(name) => {
130            return Err(qbrs_core::select::UnresolvedPlaceholder(name).into());
131        }
132        // Reachable only when a column type is on in `qbrs-core` and off
133        // here: the variant exists, the arm that binds it doesn't.
134        #[allow(unreachable_patterns)]
135        other => return Err(Error::FeatureNotEnabled(other.type_name())),
136    })
137}
138
139fn bind_all<'q>(
140    mut query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
141    params: Vec<Value>,
142) -> Result<sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>> {
143    for p in params {
144        query = bind_value(query, p)?;
145    }
146    Ok(query)
147}
148
149/// Generic over `E: sqlx::PgExecutor` so every `.load()`/`.execute()` works
150/// against a `&PgPool` or a transaction alike. sqlx implements `Executor` for
151/// `&mut PgConnection`, not `Transaction`, so callers pass `&mut *tx`.
152async fn fetch_all<'e, T: DecodeRow, E: sqlx::PgExecutor<'e>>(
153    executor: E,
154    sql: &str,
155    params: Vec<Value>,
156) -> Result<Vec<T>> {
157    let rows = bind_all(sqlx::query(sqlx::AssertSqlSafe(sql)), params)?
158        .fetch_all(executor)
159        .await?;
160    rows.iter()
161        .map(|row| T::decode_at(row, &mut 0).map_err(Error::from))
162        .collect()
163}
164
165async fn fetch_optional<'e, T: DecodeRow, E: sqlx::PgExecutor<'e>>(
166    executor: E,
167    sql: &str,
168    params: Vec<Value>,
169) -> Result<Option<T>> {
170    let row = bind_all(sqlx::query(sqlx::AssertSqlSafe(sql)), params)?
171        .fetch_optional(executor)
172        .await?;
173    row.as_ref()
174        .map(|r| T::decode_at(r, &mut 0).map_err(Error::from))
175        .transpose()
176}
177
178/// The streaming counterpart: rows are decoded as they arrive rather than
179/// collected first. The query owns its SQL and its binds, so the stream
180/// borrows only the executor.
181fn fetch_stream<'e, T: DecodeRow + Send + 'e, E: sqlx::PgExecutor<'e>>(
182    executor: E,
183    sql: String,
184    params: Vec<Value>,
185) -> Result<impl Stream<Item = Result<T>> + Send + Unpin + 'e> {
186    let query = bind_all(sqlx::query(sqlx::AssertSqlSafe(sql)), params)?;
187    Ok(futures_util::StreamExt::map(query.fetch(executor), |row| {
188        T::decode_at(&row?, &mut 0).map_err(Error::from)
189    }))
190}
191
192async fn execute_only<'e, E: sqlx::PgExecutor<'e>>(
193    executor: E,
194    sql: &str,
195    params: Vec<Value>,
196) -> Result<u64> {
197    let result = bind_all(sqlx::query(sqlx::AssertSqlSafe(sql)), params)?
198        .execute(executor)
199        .await?;
200    Ok(result.rows_affected())
201}
202
203/// What a row-producing query renders to, and what its rows decode to: a
204/// `SELECT`, a `RETURNING` clause, an erased `DynSelect`, a `UNION` chain.
205/// `LoadExt` is the methods over it, and the split is load-bearing. With
206/// the validity bound on the impl instead, an invalid selection makes
207/// `.load(..)` not exist, and the scope error the builder wanted to report
208/// is replaced by a method-resolution failure that never mentions the table.
209///
210/// `Idx` is threaded through the trait's parameter list for the reason
211/// `scope::Superset` explains. Callers never see it; it's inferred.
212#[diagnostic::on_unimplemented(
213    message = "`{Self}` isn't a query this crate can run",
214    label = "a `Select`, a `RETURNING`, a `DynSelect`, a set operation, or a `SELECT` with no `FROM`, in the `Postgres` dialect, whose values are all types `DecodeRow` covers"
215)]
216pub trait RowQuery<Idx> {
217    type Output: DecodeRow;
218
219    #[doc(hidden)]
220    fn rendered(&self) -> (String, Vec<Value>);
221}
222
223/// `load` for the rows, `load_one` for the first of them, `stream` for
224/// them one at a time, and `ExecuteExt::execute` where there are none to
225/// decode. One trait for every row-producing builder keeps the terminal
226/// vocabulary tied to what a statement yields rather than to which builder
227/// happens to be in hand.
228///
229/// Implemented for every builder, satisfiable by the ones that produce
230/// rows: what a builder can't do is then reported by `RowQuery`, which says
231/// so, rather than by the method not existing, which rustc answers with a
232/// list of unsatisfied bounds or, worse, by suggesting `Iterator`. Not a
233/// blanket impl, since `load`/`count`/`execute` are names other traits in a
234/// caller's scope have too. A builder added here needs its three empty
235/// impls, or its terminal goes back to reporting nothing.
236pub trait LoadExt {
237    fn load<'e, Idx, E: sqlx::PgExecutor<'e>>(
238        &self,
239        executor: E,
240    ) -> impl std::future::Future<Output = Result<Vec<<Self as RowQuery<Idx>>::Output>>>
241    where
242        Self: RowQuery<Idx>,
243    {
244        let (sql, params) = self.rendered();
245        async move { fetch_all::<<Self as RowQuery<Idx>>::Output, E>(executor, &sql, params).await }
246    }
247
248    fn load_one<'e, Idx, E: sqlx::PgExecutor<'e>>(
249        &self,
250        executor: E,
251    ) -> impl std::future::Future<Output = Result<Option<<Self as RowQuery<Idx>>::Output>>>
252    where
253        Self: RowQuery<Idx>,
254    {
255        let (sql, params) = self.rendered();
256        async move { fetch_optional::<<Self as RowQuery<Idx>>::Output, E>(executor, &sql, params).await }
257    }
258
259    /// The rows one at a time, for a result too large to hold: an export
260    /// that writes as it reads rather than collecting first. Yields the
261    /// same values `.load(..)` would, decoded as each row arrives.
262    ///
263    /// Not a cursor: the server still produces the whole result, and the
264    /// connection is held until the stream is dropped or exhausted. What
265    /// this bounds is the client's memory, which is what a `Vec` of every
266    /// row costs.
267    ///
268    /// Returns a `Result` around the stream rather than as its first item,
269    /// because what can fail before a row arrives (an unresolved
270    /// placeholder, a value whose feature is on in `qbrs` and off here)
271    /// is a misuse of this crate rather than a row that didn't decode.
272    /// Everything the database has to say arrives as an item.
273    ///
274    /// `Send` and `Unpin` are promised, so the stream can be spawned and
275    /// polled without pinning it first, which a generic caller has no way to
276    /// ask for otherwise.
277    fn stream<'e, Idx, E: sqlx::PgExecutor<'e>>(
278        &self,
279        executor: E,
280    ) -> Result<impl Stream<Item = Result<<Self as RowQuery<Idx>>::Output>> + Send + Unpin + 'e>
281    where
282        Self: RowQuery<Idx>,
283        <Self as RowQuery<Idx>>::Output: Send + 'e,
284    {
285        let (sql, params) = self.rendered();
286        fetch_stream::<<Self as RowQuery<Idx>>::Output, E>(executor, sql, params)
287    }
288}
289
290impl<D, Scope, Sel, Outer> LoadExt for Select<D, Scope, Sel, Outer> {}
291impl<Sel> LoadExt for qbrs_core::select::SelectSeed<Sel> {}
292impl<S, Sel> LoadExt for Returning<S, Sel> {}
293impl<D, Output> LoadExt for DynSelect<D, Output> {}
294impl<D, Output> LoadExt for SetOp<D, Output> {}
295impl<D, R: qbrs_core::insert::InsertRow> LoadExt for Insert<D, R> {}
296impl<D, T: qbrs_core::scope::Table> LoadExt for qbrs_core::insert::InsertSelect<D, T> {}
297impl<D, T: qbrs_core::scope::Table> LoadExt for Update<D, T> {}
298impl<D, T: qbrs_core::scope::Table> LoadExt for Delete<D, T> {}
299
300impl<Scope, Sel, Idx> RowQuery<Idx> for Select<Postgres, Scope, Sel>
301where
302    Sel: Selection<Scope, Idx>,
303    Sel::Output: DecodeRow,
304{
305    type Output = Sel::Output;
306
307    fn rendered(&self) -> (String, Vec<Value>) {
308        self.to_sql::<Idx>(Postgres)
309    }
310}
311
312/// A `SELECT` with no `FROM`: the seed is the whole statement, so it is
313/// what carries the terminal.
314impl<Sel, Idx> RowQuery<Idx> for qbrs_core::select::SelectSeed<Sel>
315where
316    Sel: Selection<qbrs_core::scope::Nil, Idx>,
317    Sel::Output: DecodeRow,
318{
319    type Output = Sel::Output;
320
321    fn rendered(&self) -> (String, Vec<Value>) {
322        self.to_sql::<Postgres, Idx>(Postgres)
323    }
324}
325
326/// `SELECT count(*)` over a query's `FROM`/`JOIN`/`WHERE`/`GROUP BY`, with
327/// its `ORDER BY`/`LIMIT`/`OFFSET` dropped: a total counts the rows that
328/// match, not the page being shown. Returns a number rather than an
329/// `Option`, since a count query always produces exactly one row.
330#[diagnostic::on_unimplemented(
331    message = "`{Self}` isn't a query this crate can count",
332    label = "a `Select`, a `DynSelect` or a set operation in the `Postgres` dialect is; a writing statement reports rows affected through `.execute(..)` instead. A `SELECT` with no `FROM` returns one row, so this error usually means a `.from(..)` was left off"
333)]
334pub trait CountQuery<Idx> {
335    #[doc(hidden)]
336    fn count_rendered(&self) -> (String, Vec<Value>);
337}
338
339/// The bound is on the method, and the impls are per-builder, for the two
340/// reasons `LoadExt` explains.
341pub trait CountExt {
342    fn count<'e, Idx, E: sqlx::PgExecutor<'e>>(
343        &self,
344        executor: E,
345    ) -> impl std::future::Future<Output = Result<i64>>
346    where
347        Self: CountQuery<Idx>,
348    {
349        count_rows(executor, self.count_rendered())
350    }
351}
352
353impl<D, Scope, Sel, Outer> CountExt for Select<D, Scope, Sel, Outer> {}
354impl<Sel> CountExt for qbrs_core::select::SelectSeed<Sel> {}
355impl<S, Sel> CountExt for Returning<S, Sel> {}
356impl<D, Output> CountExt for DynSelect<D, Output> {}
357impl<D, Output> CountExt for SetOp<D, Output> {}
358impl<D, R: qbrs_core::insert::InsertRow> CountExt for Insert<D, R> {}
359impl<D, T: qbrs_core::scope::Table> CountExt for qbrs_core::insert::InsertSelect<D, T> {}
360impl<D, T: qbrs_core::scope::Table> CountExt for Update<D, T> {}
361impl<D, T: qbrs_core::scope::Table> CountExt for Delete<D, T> {}
362
363impl<Scope, Sel: Selection<Scope, Idx>, Idx> CountQuery<Idx> for Select<Postgres, Scope, Sel> {
364    fn count_rendered(&self) -> (String, Vec<Value>) {
365        self.count_sql::<Idx>(Postgres)
366    }
367}
368
369/// Erasure is for a query whose joins depend on a condition, and such a
370/// query is paged like any other, so it counts like any other. The same
371/// goes for a set-operation chain.
372impl<Output> CountQuery<()> for DynSelect<Postgres, Output> {
373    fn count_rendered(&self) -> (String, Vec<Value>) {
374        self.count_sql(Postgres)
375    }
376}
377
378impl<Output> CountQuery<()> for SetOp<Postgres, Output> {
379    fn count_rendered(&self) -> (String, Vec<Value>) {
380        self.count_sql(Postgres)
381    }
382}
383
384async fn count_rows<'e, E: sqlx::PgExecutor<'e>>(
385    executor: E,
386    (sql, params): (String, Vec<Value>),
387) -> Result<i64> {
388    let row = bind_all(sqlx::query(sqlx::AssertSqlSafe(sql.as_str())), params)?
389        .fetch_one(executor)
390        .await?;
391    Ok(row.try_get::<i64, _>(0)?)
392}
393
394/// Every writing statement, rendered: what `execute` returns is rows
395/// affected, whichever of the three it was.
396#[diagnostic::on_unimplemented(
397    message = "`{Self}` isn't a statement this crate can execute",
398    label = "an `INSERT`, `UPDATE` or `DELETE` in the `Postgres` dialect is; a `SELECT` or a `RETURNING` yields rows, so it goes through `.load(..)`, and a prepared query through `.load(.., params)`"
399)]
400pub trait WriteStatement {
401    #[doc(hidden)]
402    fn write_rendered(&self) -> (String, Vec<Value>);
403}
404
405#[diagnostic::do_not_recommend]
406impl<S: Statement<Dialect = Postgres>> WriteStatement for S {
407    fn write_rendered(&self) -> (String, Vec<Value>) {
408        self.to_sql(Postgres)
409    }
410}
411
412/// The bound is on the method, and the impls are per-builder, for the two
413/// reasons `LoadExt` explains.
414pub trait ExecuteExt {
415    fn execute<'e, E: sqlx::PgExecutor<'e>>(
416        &self,
417        executor: E,
418    ) -> impl std::future::Future<Output = Result<u64>>
419    where
420        Self: WriteStatement,
421    {
422        let (sql, params) = self.write_rendered();
423        async move { execute_only(executor, &sql, params).await }
424    }
425}
426
427impl<D, Scope, Sel, Outer> ExecuteExt for Select<D, Scope, Sel, Outer> {}
428impl<Sel> ExecuteExt for qbrs_core::select::SelectSeed<Sel> {}
429impl<S, Sel> ExecuteExt for Returning<S, Sel> {}
430impl<D, Output> ExecuteExt for DynSelect<D, Output> {}
431impl<D, Output> ExecuteExt for SetOp<D, Output> {}
432impl<D, R: qbrs_core::insert::InsertRow> ExecuteExt for Insert<D, R> {}
433impl<D, T: qbrs_core::scope::Table> ExecuteExt for qbrs_core::insert::InsertSelect<D, T> {}
434impl<D, T: qbrs_core::scope::Table> ExecuteExt for Update<D, T> {}
435impl<D, T: qbrs_core::scope::Table> ExecuteExt for Delete<D, T> {}
436
437/// One impl for every `RETURNING`: what a statement returns is decided by
438/// its selection, not by which statement it was.
439impl<S: Statement<Dialect = Postgres>, Sel, Idx> RowQuery<Idx> for Returning<S, Sel>
440where
441    Sel: Selection<WrittenTable<S::Table>, Idx>,
442    Sel::Output: DecodeRow,
443{
444    type Output = Sel::Output;
445    fn rendered(&self) -> (String, Vec<Value>) {
446        self.to_sql(Postgres)
447    }
448}
449
450/// Decodes a query's `Output` positionally out of a `PgRow`. Keyed on the
451/// plain-Rust type a selection produces rather than on the selection
452/// itself: erasure leaves only `Output`, with no `Selection` impl left to
453/// hang decoding off, so this is implemented directly against the closed set
454/// of native types.
455#[diagnostic::on_unimplemented(
456    message = "`{Self}` isn't a value this crate can decode",
457    label = "every selected column has to decode to one of the featureless natives, or to a type whose feature is on here as well as on `qbrs`",
458    note = "`chrono`/`uuid`/`decimal`/`json` have to be enabled on `qbrs-sqlx` too: they are separate `cfg`s over one `Value`"
459)]
460pub trait DecodeRow: Sized {
461    #[doc(hidden)]
462    fn decode_at(row: &PgRow, idx: &mut usize) -> sqlx::Result<Self>;
463}
464
465macro_rules! decode_row_leaf {
466    ($ty:ty) => {
467        impl DecodeRow for $ty {
468            fn decode_at(row: &PgRow, idx: &mut usize) -> sqlx::Result<Self> {
469                let v = row.try_get::<$ty, _>(*idx)?;
470                *idx += 1;
471                Ok(v)
472            }
473        }
474        impl DecodeRow for Option<$ty> {
475            fn decode_at(row: &PgRow, idx: &mut usize) -> sqlx::Result<Self> {
476                let v = row.try_get::<Option<$ty>, _>(*idx)?;
477                *idx += 1;
478                Ok(v)
479            }
480        }
481    };
482}
483decode_row_leaf!(i32);
484decode_row_leaf!(i64);
485decode_row_leaf!(f64);
486decode_row_leaf!(String);
487decode_row_leaf!(bool);
488decode_row_leaf!(Vec<u8>);
489decode_row_leaf!(Vec<String>);
490decode_row_leaf!(Vec<i32>);
491decode_row_leaf!(Vec<i64>);
492#[cfg(feature = "uuid")]
493decode_row_leaf!(Vec<uuid::Uuid>);
494#[cfg(feature = "json")]
495decode_row_leaf!(serde_json::Value);
496#[cfg(feature = "chrono")]
497decode_row_leaf!(chrono::DateTime<chrono::Utc>);
498#[cfg(feature = "chrono")]
499decode_row_leaf!(chrono::NaiveDate);
500#[cfg(feature = "uuid")]
501decode_row_leaf!(uuid::Uuid);
502#[cfg(feature = "decimal")]
503decode_row_leaf!(rust_decimal::Decimal);
504
505impl DecodeRow for RowNil {
506    fn decode_at(_row: &PgRow, _idx: &mut usize) -> sqlx::Result<Self> {
507        Ok(RowNil)
508    }
509}
510
511impl<K, V: DecodeRow, Tail: DecodeRow> DecodeRow for RowCons<K, V, Tail> {
512    fn decode_at(row: &PgRow, idx: &mut usize) -> sqlx::Result<Self> {
513        let value = V::decode_at(row, idx)?;
514        Ok(RowCons::new(value, Tail::decode_at(row, idx)?))
515    }
516}
517
518impl<L: DecodeRow> DecodeRow for Row<L> {
519    fn decode_at(row: &PgRow, idx: &mut usize) -> sqlx::Result<Self> {
520        Ok(Row::new(L::decode_at(row, idx)?))
521    }
522}
523
524/// An erased query and a set-op chain were both rendered before their
525/// selection type was gone, leaving nothing for `Idx` to index. Hence
526/// `RowQuery<()>`, the same trait with an empty proof.
527impl<Output: DecodeRow> RowQuery<()> for DynSelect<Postgres, Output> {
528    type Output = Output;
529    fn rendered(&self) -> (String, Vec<Value>) {
530        self.to_sql(Postgres)
531    }
532}
533
534impl<Output: DecodeRow> RowQuery<()> for SetOp<Postgres, Output> {
535    type Output = Output;
536    fn rendered(&self) -> (String, Vec<Value>) {
537        self.to_sql(Postgres)
538    }
539}
540
541/// Runs a `prepare!{}`-built query, resolving its named placeholders from
542/// `params` first. Separate from `LoadExt` only because the values arrive at
543/// the call rather than being baked into the query: one `Prepared` is meant
544/// to serve many calls, and `.resolve()` clones the template rather than
545/// re-rendering it.
546#[diagnostic::on_unimplemented(
547    message = "`{Self}` isn't a prepared query this crate can run",
548    label = "a `.prepare()`-built query is `Prepared<D, Params, Output>` (params before output), and its `Params` have to be the ones it declared"
549)]
550pub trait PreparedQuery<Params> {
551    type Output: DecodeRow;
552    #[doc(hidden)]
553    fn resolved(&self, params: Params) -> Result<(String, Vec<Value>)>;
554}
555
556/// The bound is on the method, and the impls are per-builder, for the two
557/// reasons `LoadExt` explains.
558pub trait PreparedExt {
559    fn load<'e, Params, E: sqlx::PgExecutor<'e>>(
560        &self,
561        executor: E,
562        params: Params,
563    ) -> impl std::future::Future<Output = Result<Vec<<Self as PreparedQuery<Params>>::Output>>>
564    where
565        Self: PreparedQuery<Params>,
566    {
567        let resolved = self.resolved(params);
568        async move {
569            let (sql, values) = resolved?;
570            fetch_all::<<Self as PreparedQuery<Params>>::Output, E>(executor, &sql, values).await
571        }
572    }
573
574    fn load_one<'e, Params, E: sqlx::PgExecutor<'e>>(
575        &self,
576        executor: E,
577        params: Params,
578    ) -> impl std::future::Future<Output = Result<Option<<Self as PreparedQuery<Params>>::Output>>>
579    where
580        Self: PreparedQuery<Params>,
581    {
582        let resolved = self.resolved(params);
583        async move {
584            let (sql, values) = resolved?;
585            fetch_optional::<<Self as PreparedQuery<Params>>::Output, E>(executor, &sql, values)
586                .await
587        }
588    }
589
590    /// The rows one at a time, as `LoadExt::stream` gives them, with the
591    /// `Params` that arrive at the call. That is what a reusable export
592    /// query wants.
593    fn stream<'e, Params, E: sqlx::PgExecutor<'e>>(
594        &self,
595        executor: E,
596        params: Params,
597    ) -> Result<
598        impl Stream<Item = Result<<Self as PreparedQuery<Params>>::Output>> + Send + Unpin + 'e,
599    >
600    where
601        Self: PreparedQuery<Params>,
602        <Self as PreparedQuery<Params>>::Output: Send + 'e,
603    {
604        let (sql, values) = self.resolved(params)?;
605        fetch_stream::<<Self as PreparedQuery<Params>>::Output, E>(executor, sql, values)
606    }
607}
608
609impl<D, Params, Output> PreparedExt for Prepared<D, Params, Output> {}
610
611// A prepared query's `load`/`count` are told apart from the plain ones by
612// arity, but `execute` is not. Without this impl, it is the one terminal on
613// the one builder that reports nothing.
614impl<D, Params, Output> ExecuteExt for Prepared<D, Params, Output> {}
615
616#[diagnostic::do_not_recommend]
617impl<Params: PreparedParams, Output: DecodeRow> PreparedQuery<Params>
618    for Prepared<Postgres, Params, Output>
619{
620    type Output = Output;
621
622    fn resolved(&self, params: Params) -> Result<(String, Vec<Value>)> {
623        Ok(self.resolve(params)?)
624    }
625}
626
627/// A prepared total. Separate from `PreparedExt` for the reason `CountExt`
628/// is separate from `LoadExt`: a count produces a number, not rows.
629#[diagnostic::on_unimplemented(
630    message = "`{Self}` isn't a prepared total this crate can run",
631    label = "`.prepare_count()` builds one; `.prepare()` builds a query whose rows go through `.load(..)`"
632)]
633pub trait PreparedTotal<Params> {
634    #[doc(hidden)]
635    fn resolved_count(&self, params: Params) -> Result<(String, Vec<Value>)>;
636}
637
638impl<Params: PreparedParams> PreparedTotal<Params> for Prepared<Postgres, Params, Total> {
639    fn resolved_count(&self, params: Params) -> Result<(String, Vec<Value>)> {
640        Ok(self.resolve(params)?)
641    }
642}
643
644/// The bound is on the method, and the impls are per-builder, for the two
645/// reasons `LoadExt` explains.
646pub trait PreparedCountExt {
647    fn count<'e, Params, E: sqlx::PgExecutor<'e>>(
648        &self,
649        executor: E,
650        params: Params,
651    ) -> impl std::future::Future<Output = Result<i64>>
652    where
653        Self: PreparedTotal<Params>,
654    {
655        let resolved = self.resolved_count(params);
656        async move { count_rows(executor, resolved?).await }
657    }
658}
659
660impl<D, Params, Output> PreparedCountExt for Prepared<D, Params, Output> {}
661
662#[cfg(test)]
663mod tests {
664    use super::*;
665
666    // No real Postgres needed: binding inspects the `Value` enum before
667    // anything reaches the network, so an unresolved placeholder is
668    // reachable, and its error checkable, without a live DB.
669    #[test]
670    fn unresolved_placeholder_is_a_typed_error_not_a_sqlx_configuration_string() {
671        let query = sqlx::query(sqlx::AssertSqlSafe("SELECT $1"));
672        let err = match bind_all(query, vec![Value::Placeholder("email")]) {
673            Err(e) => e,
674            Ok(_) => panic!("unresolved placeholder must fail to bind"),
675        };
676
677        assert!(matches!(
678            err,
679            Error::UnresolvedPlaceholder(qbrs_core::select::UnresolvedPlaceholder("email"))
680        ));
681        // A real `std::error::Error`, so it composes with
682        // `anyhow`/`Box<dyn Error>`.
683        let _: &dyn std::error::Error = &err;
684        assert_eq!(err.to_string(), "no value provided for placeholder `email`");
685    }
686
687    #[test]
688    fn sqlx_errors_convert_via_from() {
689        let err: Error = sqlx::Error::RowNotFound.into();
690        assert!(matches!(err, Error::Sqlx(sqlx::Error::RowNotFound)));
691    }
692}
qbrs_sqlx/lib.rs

qbrs_sqlx/
lib.rs
