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

persistant_sqlx/
lib.rs
