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

persistant_sqlx/
lib.rs
