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

persistant_sqlx/
lib.rs
