persistant_sqlx/
lib.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! PersistAnt's relational driver over sqlx (PostgreSQL first).
3//!
4//! [`PgBackend`] implements [`persistant_relational::RelationalBackend`]. The concrete
5//! [`sqlx::PgPool`] is available through [`PgBackend::pool`], so a program that wants
6//! compile-time-checked `sqlx::query!` keeps it (D1, C+): this crate does not hide sqlx.
7//!
8//! Failures are classified from the server's SQLSTATE into [`persistant_relational::RelError`].
9
10mod bounded;
11#[cfg(feature = "bridge")]
12mod bridge;
13mod decode;
14mod migrate;
15#[cfg(feature = "mysql")]
16mod mysql;
17mod options;
18#[cfg(feature = "sqlite")]
19mod sqlite;
20mod testdb;
21
22pub use bounded::{Bounded, BoundedTx, call_bound_from_env, parse_bound};
23#[cfg(feature = "bridge")]
24pub use bridge::{OnConn, OnPool};
25pub use migrate::{FromSqlxError, migrations_from_sqlx};
26#[cfg(feature = "mysql")]
27pub use mysql::{MysqlBackend, MysqlTx};
28pub use options::{ConnectOptions, TlsPolicy};
29pub use persistant_relational::{MigrateError, MigrateReport, Migrates, Migration};
30#[cfg(feature = "sqlite")]
31pub use sqlite::{SqliteBackend, SqliteTx};
32pub use testdb::TestDb;
33
34use persistant_relational::{
35    Dialect, Isolation, RelError, RelationalBackend, RelationalTx, Row, Value,
36};
37use sqlx::postgres::{PgArguments, PgPoolOptions, PgRow};
38use sqlx::query::Query;
39use sqlx::{Column, Postgres, Row as SqlxRow, TypeInfo, ValueRef};
40
41/// Classify an sqlx error by SQLSTATE (https://www.postgresql.org/docs/current/errcodes-appendix.html).
42pub(crate) fn classify(e: sqlx::Error) -> RelError {
43    match &e {
44        sqlx::Error::Database(db) => {
45            let constraint = db.constraint().map(str::to_string);
46            match db.code().as_deref() {
47                Some("23505") => RelError::UniqueViolation { constraint },
48                Some("23503") => RelError::ForeignKeyViolation { constraint },
49                Some("23514") | Some("23502") => RelError::CheckViolation { constraint },
50                Some("40001") => RelError::Serialization,
51                Some("40P01") => RelError::Deadlock,
52                Some(c) if c.starts_with("08") || c.starts_with("57P") => {
53                    RelError::Unavailable(e.to_string())
54                }
55                _ => RelError::Backend(e.to_string()),
56            }
57        }
58        sqlx::Error::Io(_) | sqlx::Error::PoolTimedOut | sqlx::Error::PoolClosed => {
59            RelError::Unavailable(e.to_string())
60        }
61        _ => RelError::Backend(e.to_string()),
62    }
63}
64
65/// A `NULL` whose type the server infers from the statement. Binding `Option::<String>::None` sends
66/// the type `text`, which the server rejects for a `bigint` column (`expression is of type text`) the
67/// first time a pooled connection prepares that statement, and not on one that already has: a failure
68/// that depends on which connection the pool hands out. Type OID 0 means "unspecified".
69struct UntypedNull;
70
71impl sqlx::Type<Postgres> for UntypedNull {
72    fn type_info() -> sqlx::postgres::PgTypeInfo {
73        sqlx::postgres::PgTypeInfo::with_oid(sqlx::postgres::types::Oid(0))
74    }
75
76    fn compatible(_: &sqlx::postgres::PgTypeInfo) -> bool {
77        true
78    }
79}
80
81impl<'q> sqlx::Encode<'q, Postgres> for UntypedNull {
82    fn encode_by_ref(
83        &self,
84        _: &mut sqlx::postgres::PgArgumentBuffer,
85    ) -> Result<sqlx::encode::IsNull, sqlx::error::BoxDynError> {
86        Ok(sqlx::encode::IsNull::Yes)
87    }
88}
89
90fn bind<'q>(
91    mut q: Query<'q, Postgres, PgArguments>,
92    params: &'q [Value],
93) -> Result<Query<'q, Postgres, PgArguments>, RelError> {
94    for p in params {
95        q = match p {
96            // A NULL of unspecified type: the server infers it from the statement.
97            Value::Null => q.bind(UntypedNull),
98            Value::Bool(b) => q.bind(*b),
99            Value::Int(n) => q.bind(*n),
100            Value::Float(f) => q.bind(*f),
101            Value::Text(s) => q.bind(s.as_str()),
102            Value::Bytes(b) => q.bind(b.as_slice()),
103            other => {
104                return Err(RelError::Unsupported(format!(
105                    "parameter {other:?} has no PostgreSQL binding in this driver"
106                )));
107            }
108        };
109    }
110    Ok(q)
111}
112
113fn decode_failed(e: sqlx::error::BoxDynError) -> RelError {
114    RelError::Backend(e.to_string())
115}
116
117fn malformed(column: &str, ty: &str, why: String) -> RelError {
118    RelError::Backend(format!("column {column:?} ({ty}): {why}"))
119}
120
121fn decode(row: &PgRow) -> Result<Row, RelError> {
122    let mut values = Vec::with_capacity(row.columns().len());
123    for (i, col) in row.columns().iter().enumerate() {
124        let raw = row.try_get_raw(i).map_err(classify)?;
125        if raw.is_null() {
126            values.push(Value::Null);
127            continue;
128        }
129        let name = col.type_info().name().to_string();
130        let v = match name.as_str() {
131            "BOOL" => Value::Bool(row.try_get::<bool, _>(i).map_err(classify)?),
132            "INT2" => Value::Int(i64::from(row.try_get::<i16, _>(i).map_err(classify)?)),
133            "INT4" => Value::Int(i64::from(row.try_get::<i32, _>(i).map_err(classify)?)),
134            "INT8" => Value::Int(row.try_get::<i64, _>(i).map_err(classify)?),
135            "FLOAT4" => Value::Float(decode::f32_shortest(
136                row.try_get::<f32, _>(i).map_err(classify)?,
137            )),
138            "FLOAT8" => Value::Float(row.try_get::<f64, _>(i).map_err(classify)?),
139            // sqlx names `bpchar` (char(n)) "CHAR"; the quoted "CHAR" is the one-byte internal type.
140            "TEXT" | "VARCHAR" | "NAME" | "BPCHAR" | "CHAR" => {
141                Value::Text(row.try_get::<String, _>(i).map_err(classify)?)
142            }
143            "BYTEA" => Value::Bytes(row.try_get::<Vec<u8>, _>(i).map_err(classify)?),
144            // A text array: JSON array text, a NULL element as `null` (`FromValue for Vec<String>`).
145            "TEXT[]" | "VARCHAR[]" | "NAME[]" | "CHAR[]" => Value::Text(decode::json_string_array(
146                &row.try_get::<Vec<Option<String>>, _>(i).map_err(classify)?,
147            )),
148            // Types with a canonical text form (`decode.rs`): carried as that text.
149            "UUID" => Value::Text(
150                decode::uuid_text(raw.as_bytes().map_err(decode_failed)?)
151                    .map_err(|m| malformed(col.name(), &name, m))?,
152            ),
153            "NUMERIC" => Value::Text(
154                decode::numeric_text(raw.as_bytes().map_err(decode_failed)?)
155                    .map_err(|m| malformed(col.name(), &name, m))?,
156            ),
157            "INET" | "CIDR" => Value::Text(
158                decode::inet_text(raw.as_bytes().map_err(decode_failed)?)
159                    .map_err(|m| malformed(col.name(), &name, m))?,
160            ),
161            "JSON" => Value::Text(
162                String::from_utf8(raw.as_bytes().map_err(decode_failed)?.to_vec())
163                    .map_err(|e| malformed(col.name(), &name, e.to_string()))?,
164            ),
165            "JSONB" => {
166                // Version byte 1, then the JSON text.
167                let b = raw.as_bytes().map_err(decode_failed)?;
168                match b.split_first() {
169                    Some((1, rest)) => Value::Text(
170                        String::from_utf8(rest.to_vec())
171                            .map_err(|e| malformed(col.name(), &name, e.to_string()))?,
172                    ),
173                    _ => return Err(malformed(col.name(), &name, "unknown jsonb version".into())),
174                }
175            }
176            "DATE" => Value::Text(decode::date_text(
177                row.try_get::<sqlx::types::chrono::NaiveDate, _>(i)
178                    .map_err(classify)?,
179            )),
180            "TIME" => Value::Text(decode::time_text(
181                row.try_get::<sqlx::types::chrono::NaiveTime, _>(i)
182                    .map_err(classify)?,
183            )),
184            "TIMESTAMP" => Value::Text(decode::naive_text(
185                row.try_get::<sqlx::types::chrono::NaiveDateTime, _>(i)
186                    .map_err(classify)?,
187            )),
188            "TIMESTAMPTZ" => Value::Text(decode::utc_text(
189                row.try_get::<sqlx::types::chrono::DateTime<sqlx::types::chrono::Utc>, _>(i)
190                    .map_err(classify)?,
191            )),
192            other => {
193                return Err(RelError::Unsupported(format!(
194                    "column {:?} has type {other}, which Value cannot carry; use PgBackend::pool",
195                    col.name()
196                )));
197            }
198        };
199        values.push(v);
200    }
201    Ok(Row::new(values))
202}
203
204/// A PostgreSQL database behind an sqlx pool.
205#[derive(Debug, Clone)]
206pub struct PgBackend {
207    pool: sqlx::PgPool,
208    version: String,
209}
210
211impl PgBackend {
212    /// Connect with default pool settings and read the server version.
213    ///
214    /// Prefer [`PgBackend::connect_secret`]: a `&str` URL is one `Debug` print from a log file.
215    pub async fn connect(url: &str) -> Result<Self, RelError> {
216        let pool = PgPoolOptions::new()
217            .max_connections(16)
218            .connect(url)
219            .await
220            .map_err(classify)?;
221        Self::from_pool(pool).await
222    }
223
224    /// Connect with a connection string held as a [`persistant_relational::ConnectionSecret`]:
225    /// resolved here, never printed, never part of an error.
226    pub async fn connect_secret(
227        secret: &persistant_relational::ConnectionSecret,
228    ) -> Result<Self, RelError> {
229        let url = secret.resolve()?;
230        Self::connect(url.expose()).await
231    }
232
233    /// Wrap an existing pool (a program that already owns one keeps it).
234    pub async fn from_pool(pool: sqlx::PgPool) -> Result<Self, RelError> {
235        let full: String = sqlx::query_scalar("SHOW server_version")
236            .fetch_one(&pool)
237            .await
238            .map_err(classify)?;
239        // "16.15 (Debian 16.15-1.pgdg13+2)" -> "16.15"
240        let version = full.split_whitespace().next().unwrap_or("").to_string();
241        Ok(PgBackend { pool, version })
242    }
243
244    /// The concrete pool: for `sqlx::query!` and anything this layer does not carry.
245    pub fn pool(&self) -> &sqlx::PgPool {
246        &self.pool
247    }
248}
249
250impl RelationalBackend for PgBackend {
251    type Tx<'a> = PgTx<'a>;
252
253    fn backend(&self) -> &str {
254        "postgres"
255    }
256
257    fn version(&self) -> &str {
258        &self.version
259    }
260
261    fn dialect(&self) -> Dialect {
262        Dialect::Postgres
263    }
264
265    async fn execute(&self, sql: &str, params: &[Value]) -> Result<u64, RelError> {
266        let q = bind(sqlx::query(sqlx::AssertSqlSafe(sql.to_string())), params)?;
267        q.execute(&self.pool)
268            .await
269            .map(|r| r.rows_affected())
270            .map_err(classify)
271    }
272
273    async fn query(&self, sql: &str, params: &[Value]) -> Result<Vec<Row>, RelError> {
274        let q = bind(sqlx::query(sqlx::AssertSqlSafe(sql.to_string())), params)?;
275        let rows = q.fetch_all(&self.pool).await.map_err(classify)?;
276        rows.iter().map(decode).collect()
277    }
278
279    async fn begin(&self, isolation: Isolation) -> Result<PgTx<'_>, RelError> {
280        let statement = match isolation {
281            Isolation::ReadCommitted => "BEGIN",
282            Isolation::RepeatableRead => "BEGIN ISOLATION LEVEL REPEATABLE READ",
283            Isolation::Serializable => "BEGIN ISOLATION LEVEL SERIALIZABLE",
284            other => return Err(RelError::Unsupported(format!("isolation level {other:?}"))),
285        };
286        Ok(PgTx {
287            tx: self.pool.begin_with(statement).await.map_err(classify)?,
288        })
289    }
290}
291
292/// A PostgreSQL transaction.
293#[derive(Debug)]
294pub struct PgTx<'a> {
295    tx: sqlx::Transaction<'a, Postgres>,
296}
297
298impl RelationalTx for PgTx<'_> {
299    async fn execute(&mut self, sql: &str, params: &[Value]) -> Result<u64, RelError> {
300        let q = bind(sqlx::query(sqlx::AssertSqlSafe(sql.to_string())), params)?;
301        q.execute(&mut *self.tx)
302            .await
303            .map(|r| r.rows_affected())
304            .map_err(classify)
305    }
306
307    async fn query(&mut self, sql: &str, params: &[Value]) -> Result<Vec<Row>, RelError> {
308        let q = bind(sqlx::query(sqlx::AssertSqlSafe(sql.to_string())), params)?;
309        let rows = q.fetch_all(&mut *self.tx).await.map_err(classify)?;
310        rows.iter().map(decode).collect()
311    }
312
313    async fn commit(self) -> Result<(), RelError> {
314        self.tx.commit().await.map_err(classify)
315    }
316
317    async fn rollback(self) -> Result<(), RelError> {
318        self.tx.rollback().await.map_err(classify)
319    }
320}
persistant_sqlx/lib.rs

persistant_sqlx/
lib.rs
