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

persistant_sqlx/
lib.rs
