sea_orm/driver/
sqlx_sqlite.rs

1use futures_util::lock::Mutex;
2use log::LevelFilter;
3use sea_query::Values;
4use std::{future::Future, pin::Pin, sync::Arc};
5
6use sqlx::{
7    Connection, Executor, Sqlite, SqlitePool,
8    pool::PoolConnection,
9    sqlite::{SqliteConnectOptions, SqliteQueryResult, SqliteRow},
10};
11
12use sea_query_sqlx::SqlxValues;
13use tracing::{instrument, warn};
14
15use crate::{
16    AccessMode, ConnectOptions, DatabaseConnection, DatabaseConnectionType, DatabaseTransaction,
17    IsolationLevel, SqliteTransactionMode, Statement, TransactionError, debug_print, error::*,
18    executor::*, sqlx_error_to_exec_err,
19};
20
21use super::sqlx_common::*;
22
23#[cfg(feature = "stream")]
24use crate::QueryStream;
25
26/// Defines the [sqlx::sqlite] connector
27#[derive(Debug)]
28pub struct SqlxSqliteConnector;
29
30/// Defines a sqlx SQLite pool
31#[derive(Clone)]
32pub struct SqlxSqlitePoolConnection {
33    pub(crate) pool: SqlitePool,
34    metric_callback: Option<crate::metric::Callback>,
35    pub(crate) record_stmt_in_spans: bool,
36}
37
38impl std::fmt::Debug for SqlxSqlitePoolConnection {
39    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
40        write!(f, "SqlxSqlitePoolConnection {{ pool: {:?} }}", self.pool)
41    }
42}
43
44impl From<SqlitePool> for SqlxSqlitePoolConnection {
45    fn from(pool: SqlitePool) -> Self {
46        SqlxSqlitePoolConnection {
47            pool,
48            metric_callback: None,
49            record_stmt_in_spans: true,
50        }
51    }
52}
53
54impl From<SqlitePool> for DatabaseConnection {
55    fn from(pool: SqlitePool) -> Self {
56        DatabaseConnectionType::SqlxSqlitePoolConnection(pool.into()).into()
57    }
58}
59
60impl SqlxSqliteConnector {
61    /// Check if the URI provided corresponds to `sqlite:` for a SQLite database
62    pub fn accepts(string: &str) -> bool {
63        string.starts_with("sqlite:") && string.parse::<SqliteConnectOptions>().is_ok()
64    }
65
66    /// Add configuration options for the SQLite database
67    #[instrument(level = "trace")]
68    pub async fn connect(options: ConnectOptions) -> Result<DatabaseConnection, DbErr> {
69        let mut options = options;
70        let record_stmt_in_spans = options.get_record_stmt_in_spans();
71        let mut sqlx_opts = options
72            .url
73            .parse::<SqliteConnectOptions>()
74            .map_err(sqlx_error_to_conn_err)?;
75        if let Some(sqlcipher_key) = &options.sqlcipher_key {
76            sqlx_opts = sqlx_opts.pragma("key", sqlcipher_key.clone());
77        }
78        use sqlx::ConnectOptions;
79        if !options.sqlx_logging {
80            sqlx_opts = sqlx_opts.disable_statement_logging();
81        } else {
82            sqlx_opts = sqlx_opts.log_statements(options.sqlx_logging_level);
83            if options.sqlx_slow_statements_logging_level != LevelFilter::Off {
84                sqlx_opts = sqlx_opts.log_slow_statements(
85                    options.sqlx_slow_statements_logging_level,
86                    options.sqlx_slow_statements_logging_threshold,
87                );
88            }
89        }
90
91        if options.get_max_connections().is_none() {
92            options.max_connections(1);
93        }
94
95        if let Some(f) = &options.sqlite_opts_fn {
96            sqlx_opts = f(sqlx_opts);
97        }
98
99        let after_conn = options.after_connect.clone();
100        let connect_lazy = options.connect_lazy;
101        let sqlite_pool_opts_fn = options.sqlite_pool_opts_fn.clone();
102        let sqlite_before_acquire = options.sqlite_before_acquire_fn.clone();
103        let ping_after_idle = options.test_before_acquire_if_idle_for;
104        let mut pool_options = options.sqlx_pool_options();
105        pool_options = crate::ConnectOptions::apply_before_acquire::<sqlx::Sqlite>(
106            pool_options,
107            ping_after_idle,
108            sqlite_before_acquire,
109        );
110
111        if let Some(f) = &sqlite_pool_opts_fn {
112            pool_options = f(pool_options);
113        }
114
115        let pool = if connect_lazy {
116            pool_options.connect_lazy_with(sqlx_opts)
117        } else {
118            pool_options
119                .connect_with(sqlx_opts)
120                .await
121                .map_err(sqlx_error_to_conn_err)?
122        };
123
124        let pool = SqlxSqlitePoolConnection {
125            pool,
126            metric_callback: None,
127            record_stmt_in_spans,
128        };
129
130        #[cfg(feature = "sqlite-use-returning-for-3_35")]
131        {
132            let version = get_version(&pool).await?;
133            super::sqlite::ensure_returning_version(&version)?;
134        }
135
136        let conn: DatabaseConnection =
137            DatabaseConnectionType::SqlxSqlitePoolConnection(pool).into();
138
139        if let Some(cb) = after_conn {
140            cb(conn.clone()).await?;
141        }
142
143        Ok(conn)
144    }
145}
146
147impl SqlxSqliteConnector {
148    /// Instantiate a sqlx pool connection to a [DatabaseConnection]
149    pub fn from_sqlx_sqlite_pool(pool: SqlitePool) -> DatabaseConnection {
150        DatabaseConnectionType::SqlxSqlitePoolConnection(SqlxSqlitePoolConnection {
151            pool,
152            metric_callback: None,
153            record_stmt_in_spans: true,
154        })
155        .into()
156    }
157}
158
159impl SqlxSqlitePoolConnection {
160    /// Execute a [Statement] on a SQLite backend
161    #[instrument(level = "trace", skip(stmt))]
162    pub async fn execute(&self, stmt: Statement) -> Result<ExecResult, DbErr> {
163        debug_print!("{}", stmt);
164
165        let query = sqlx_query(&stmt);
166        let mut conn = self.pool.acquire().await.map_err(sqlx_conn_acquire_err)?;
167        crate::metric::metric!(self.metric_callback, &stmt, {
168            match query.execute(&mut *conn).await {
169                Ok(res) => Ok(res.into()),
170                Err(err) => Err(sqlx_error_to_exec_err(err)),
171            }
172        })
173    }
174
175    /// Execute an unprepared SQL statement on a SQLite backend
176    #[instrument(level = "trace", skip(sql))]
177    pub async fn execute_unprepared(&self, sql: &str) -> Result<ExecResult, DbErr> {
178        debug_print!("{}", sql);
179
180        let conn = &mut self.pool.acquire().await.map_err(sqlx_conn_acquire_err)?;
181        match conn.execute(sqlx::AssertSqlSafe(sql.to_owned())).await {
182            Ok(res) => Ok(res.into()),
183            Err(err) => Err(sqlx_error_to_exec_err(err)),
184        }
185    }
186
187    /// Get one result from a SQL query. Returns [Option::None] if no match was found
188    #[instrument(level = "trace", skip(stmt))]
189    pub async fn query_one(&self, stmt: Statement) -> Result<Option<QueryResult>, DbErr> {
190        debug_print!("{}", stmt);
191
192        let query = sqlx_query(&stmt);
193        let mut conn = self.pool.acquire().await.map_err(sqlx_conn_acquire_err)?;
194        crate::metric::metric!(self.metric_callback, &stmt, {
195            match query.fetch_one(&mut *conn).await {
196                Ok(row) => Ok(Some(row.into())),
197                Err(err) => match err {
198                    sqlx::Error::RowNotFound => Ok(None),
199                    _ => Err(sqlx_error_to_query_err(err)),
200                },
201            }
202        })
203    }
204
205    /// Get the results of a query returning them as a Vec<[QueryResult]>
206    #[instrument(level = "trace", skip(stmt))]
207    pub async fn query_all(&self, stmt: Statement) -> Result<Vec<QueryResult>, DbErr> {
208        debug_print!("{}", stmt);
209
210        let query = sqlx_query(&stmt);
211        let mut conn = self.pool.acquire().await.map_err(sqlx_conn_acquire_err)?;
212        crate::metric::metric!(self.metric_callback, &stmt, {
213            match query.fetch_all(&mut *conn).await {
214                Ok(rows) => Ok(rows.into_iter().map(|r| r.into()).collect()),
215                Err(err) => Err(sqlx_error_to_query_err(err)),
216            }
217        })
218    }
219
220    /// Stream the results of executing a SQL query
221    #[instrument(level = "trace", skip(stmt))]
222    #[cfg(feature = "stream")]
223    pub async fn stream(&self, stmt: Statement) -> Result<QueryStream, DbErr> {
224        debug_print!("{}", stmt);
225
226        let conn = self.pool.acquire().await.map_err(sqlx_conn_acquire_err)?;
227        Ok(QueryStream::from((
228            conn,
229            stmt,
230            self.metric_callback.clone(),
231        )))
232    }
233
234    /// Bundle a set of SQL statements that execute together.
235    #[instrument(level = "trace")]
236    pub async fn begin(
237        &self,
238        isolation_level: Option<IsolationLevel>,
239        access_mode: Option<AccessMode>,
240        sqlite_transaction_mode: Option<SqliteTransactionMode>,
241    ) -> Result<DatabaseTransaction, DbErr> {
242        let conn = self.pool.acquire().await.map_err(sqlx_conn_acquire_err)?;
243        DatabaseTransaction::new_sqlite(
244            conn,
245            self.metric_callback.clone(),
246            self.record_stmt_in_spans,
247            isolation_level,
248            access_mode,
249            sqlite_transaction_mode,
250        )
251        .await
252    }
253
254    /// Create a SQLite transaction
255    #[instrument(level = "trace", skip(callback))]
256    pub async fn transaction<F, T, E>(
257        &self,
258        callback: F,
259        isolation_level: Option<IsolationLevel>,
260        access_mode: Option<AccessMode>,
261    ) -> Result<T, TransactionError<E>>
262    where
263        F: for<'b> FnOnce(
264                &'b DatabaseTransaction,
265            ) -> Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'b>>
266            + Send,
267        T: Send,
268        E: std::fmt::Display + std::fmt::Debug + Send,
269    {
270        let conn = self.pool.acquire().await.map_err(sqlx_conn_acquire_err)?;
271        let transaction = DatabaseTransaction::new_sqlite(
272            conn,
273            self.metric_callback.clone(),
274            self.record_stmt_in_spans,
275            isolation_level,
276            access_mode,
277            None,
278        )
279        .await
280        .map_err(|e| TransactionError::Connection(e))?;
281        transaction.run(callback).await
282    }
283
284    pub(crate) fn set_metric_callback<F>(&mut self, callback: F)
285    where
286        F: Fn(&crate::metric::Info<'_>) + Send + Sync + 'static,
287    {
288        self.metric_callback = Some(Arc::new(callback));
289    }
290
291    /// Checks if a connection to the database is still valid.
292    pub async fn ping(&self) -> Result<(), DbErr> {
293        let conn = &mut self.pool.acquire().await.map_err(sqlx_conn_acquire_err)?;
294        match conn.ping().await {
295            Ok(_) => Ok(()),
296            Err(err) => Err(sqlx_error_to_conn_err(err)),
297        }
298    }
299
300    /// Explicitly close the SQLite connection.
301    /// See [`Self::close_by_ref`] for usage with references.
302    pub async fn close(self) -> Result<(), DbErr> {
303        self.close_by_ref().await
304    }
305
306    /// Explicitly close the SQLite connection
307    pub async fn close_by_ref(&self) -> Result<(), DbErr> {
308        self.pool.close().await;
309        Ok(())
310    }
311}
312
313impl From<SqliteRow> for QueryResult {
314    fn from(row: SqliteRow) -> QueryResult {
315        QueryResult {
316            row: QueryResultRow::SqlxSqlite(row),
317        }
318    }
319}
320
321impl From<SqliteQueryResult> for ExecResult {
322    fn from(result: SqliteQueryResult) -> ExecResult {
323        ExecResult {
324            result: ExecResultHolder::SqlxSqlite(result),
325        }
326    }
327}
328
329pub(crate) fn sqlx_query(stmt: &Statement) -> sqlx::query::Query<'_, Sqlite, SqlxValues> {
330    let values = stmt
331        .values
332        .as_ref()
333        .map_or(Values(Vec::new()), |values| values.clone());
334    sqlx::query_with(sqlx::AssertSqlSafe(stmt.sql.as_str()), SqlxValues(values))
335}
336
337pub(crate) async fn set_transaction_config(
338    _conn: &mut PoolConnection<Sqlite>,
339    isolation_level: Option<IsolationLevel>,
340    access_mode: Option<AccessMode>,
341) -> Result<(), DbErr> {
342    if isolation_level.is_some() {
343        warn!("Setting isolation level in a SQLite transaction isn't supported");
344    }
345    if access_mode.is_some() {
346        warn!("Setting access mode in a SQLite transaction isn't supported");
347    }
348    Ok(())
349}
350
351#[cfg(feature = "sqlite-use-returning-for-3_35")]
352async fn get_version(conn: &SqlxSqlitePoolConnection) -> Result<String, DbErr> {
353    let stmt = Statement {
354        sql: "SELECT sqlite_version()".to_string(),
355        values: None,
356        db_backend: crate::DbBackend::Sqlite,
357    };
358    conn.query_one(stmt)
359        .await?
360        .ok_or_else(|| {
361            DbErr::Conn(RuntimeErr::Internal(
362                "Error reading SQLite version".to_string(),
363            ))
364        })?
365        .try_get_by(0)
366}
367
368#[cfg(feature = "stream")]
369impl
370    From<(
371        PoolConnection<sqlx::Sqlite>,
372        Statement,
373        Option<crate::metric::Callback>,
374    )> for crate::QueryStream
375{
376    fn from(
377        (conn, stmt, metric_callback): (
378            PoolConnection<sqlx::Sqlite>,
379            Statement,
380            Option<crate::metric::Callback>,
381        ),
382    ) -> Self {
383        crate::QueryStream::build(stmt, crate::InnerConnection::Sqlite(conn), metric_callback)
384    }
385}
386
387impl crate::DatabaseTransaction {
388    pub(crate) async fn new_sqlite(
389        inner: PoolConnection<sqlx::Sqlite>,
390        metric_callback: Option<crate::metric::Callback>,
391        record_stmt_in_spans: bool,
392        isolation_level: Option<IsolationLevel>,
393        access_mode: Option<AccessMode>,
394        sqlite_transaction_mode: Option<SqliteTransactionMode>,
395    ) -> Result<crate::DatabaseTransaction, DbErr> {
396        Self::begin(
397            Arc::new(Mutex::new(crate::InnerConnection::Sqlite(inner))),
398            crate::DbBackend::Sqlite,
399            metric_callback,
400            record_stmt_in_spans,
401            isolation_level,
402            access_mode,
403            sqlite_transaction_mode,
404        )
405        .await
406    }
407}
408
409#[cfg(feature = "proxy")]
410pub(crate) fn from_sqlx_sqlite_row_to_proxy_row(row: &sqlx::sqlite::SqliteRow) -> crate::ProxyRow {
411    // https://docs.rs/sqlx-sqlite/0.7.2/src/sqlx_sqlite/type_info.rs.html
412    // https://docs.rs/sqlx-sqlite/0.7.2/sqlx_sqlite/types/index.html
413    use sea_query::Value;
414    use sqlx::{Column, Row, TypeInfo};
415    crate::ProxyRow {
416        values: row
417            .columns()
418            .iter()
419            .map(|c| {
420                (
421                    c.name().to_string(),
422                    match c.type_info().name() {
423                        "BOOLEAN" => {
424                            Value::Bool(row.try_get(c.ordinal()).expect("Failed to get boolean"))
425                        }
426
427                        "INTEGER" => {
428                            Value::Int(row.try_get(c.ordinal()).expect("Failed to get integer"))
429                        }
430
431                        "BIGINT" | "INT8" => Value::BigInt(
432                            row.try_get(c.ordinal()).expect("Failed to get big integer"),
433                        ),
434
435                        "REAL" => {
436                            Value::Double(row.try_get(c.ordinal()).expect("Failed to get double"))
437                        }
438
439                        "TEXT" => Value::String(
440                            row.try_get::<Option<String>, _>(c.ordinal())
441                                .expect("Failed to get string")
442                                .map(Box::new),
443                        ),
444
445                        "BLOB" => Value::Bytes(
446                            row.try_get::<Option<Vec<u8>>, _>(c.ordinal())
447                                .expect("Failed to get bytes")
448                                .map(Box::new),
449                        ),
450
451                        #[cfg(feature = "with-chrono")]
452                        "DATETIME" => {
453                            use chrono::{DateTime, Utc};
454
455                            Value::ChronoDateTimeUtc(
456                                row.try_get::<Option<DateTime<Utc>>, _>(c.ordinal())
457                                    .expect("Failed to get timestamp")
458                                    .map(Box::new),
459                            )
460                        }
461                        #[cfg(all(feature = "with-time", not(feature = "with-chrono")))]
462                        "DATETIME" => {
463                            use time::OffsetDateTime;
464                            Value::TimeDateTimeWithTimeZone(
465                                row.try_get::<Option<OffsetDateTime>, _>(c.ordinal())
466                                    .expect("Failed to get timestamp")
467                                    .map(Box::new),
468                            )
469                        }
470                        #[cfg(feature = "with-chrono")]
471                        "DATE" => {
472                            use chrono::NaiveDate;
473                            Value::ChronoDate(
474                                row.try_get::<Option<NaiveDate>, _>(c.ordinal())
475                                    .expect("Failed to get date")
476                                    .map(Box::new),
477                            )
478                        }
479                        #[cfg(all(feature = "with-time", not(feature = "with-chrono")))]
480                        "DATE" => {
481                            use time::Date;
482                            Value::TimeDate(
483                                row.try_get::<Option<Date>, _>(c.ordinal())
484                                    .expect("Failed to get date")
485                                    .map(Box::new),
486                            )
487                        }
488
489                        #[cfg(feature = "with-chrono")]
490                        "TIME" => {
491                            use chrono::NaiveTime;
492                            Value::ChronoTime(
493                                row.try_get::<Option<NaiveTime>, _>(c.ordinal())
494                                    .expect("Failed to get time")
495                                    .map(Box::new),
496                            )
497                        }
498                        #[cfg(all(feature = "with-time", not(feature = "with-chrono")))]
499                        "TIME" => {
500                            use time::Time;
501                            Value::TimeTime(
502                                row.try_get::<Option<Time>, _>(c.ordinal())
503                                    .expect("Failed to get time")
504                                    .map(Box::new),
505                            )
506                        }
507
508                        _ => unreachable!("Unknown column type: {}", c.type_info().name()),
509                    },
510                )
511            })
512            .collect(),
513    }
514}
sea_orm/driver/sqlx_sqlite.rs

sea_orm/driver/
sqlx_sqlite.rs
