dinoco_engine/databases/sqlite/
mod.rs

1use async_trait::async_trait;
2
3use std::future::Future;
4use std::pin::Pin;
5use std::sync::Arc;
6use std::time::Instant;
7
8use deadpool_sqlite::{Config, Pool, Runtime};
9use rusqlite::types::{ToSqlOutput, Value};
10
11mod dialect;
12mod handler;
13mod migration;
14mod row;
15
16use crate::{
17    ConstraintError, DinocoAdapter, DinocoClientConfig, DinocoError, DinocoQueryLog, DinocoQueryLogger, DinocoResult,
18    DinocoRow, DinocoTransactionAdapter, DinocoValue, ExecutionResult,
19};
20
21pub use dialect::SqliteDialect;
22
23static SQLITE_DIALECT: SqliteDialect = SqliteDialect;
24tokio::task_local! {
25    static SQLITE_TX_CONNECTION: Arc<tokio::sync::Mutex<deadpool_sqlite::Object>>;
26}
27
28#[derive(Clone)]
29pub struct SqliteAdapter {
30    pub url: String,
31    pub pool: Arc<Pool>,
32    pub query_logger: DinocoQueryLogger,
33}
34
35#[async_trait]
36impl DinocoAdapter for SqliteAdapter {
37    type Dialect = SqliteDialect;
38
39    fn dialect(&self) -> &Self::Dialect {
40        &SQLITE_DIALECT
41    }
42
43    async fn connect(url: String, config: DinocoClientConfig) -> DinocoResult<Self> {
44        if !url.starts_with("file:") {
45            return Err(DinocoError::ConnectionError(
46                "Invalid sqlite connection URL. Expected format: file:path".to_string(),
47            ));
48        }
49
50        let cfg = Config::new(&url);
51        let pool = cfg.create_pool(Runtime::Tokio1).map_err(DinocoError::from)?;
52
53        Ok(Self { url, pool: Arc::new(pool), query_logger: config.query_logger })
54    }
55
56    async fn execute_result(&self, query: &str, params: &[DinocoValue]) -> DinocoResult<ExecutionResult> {
57        if let Ok(tx_conn) = SQLITE_TX_CONNECTION.try_with(Clone::clone) {
58            let conn = tx_conn.lock().await;
59
60            return execute_result_with_connection(&conn, query, params, &self.query_logger).await;
61        }
62
63        let conn = self.pool.get().await.map_err(DinocoError::from)?;
64
65        execute_result_with_connection(&conn, query, params, &self.query_logger).await
66    }
67
68    async fn execute_script(&self, sql_content: &str) -> DinocoResult<()> {
69        for statement in sql_content.split(';') {
70            let clean_statement = statement.trim();
71
72            if clean_statement.is_empty() {
73                continue;
74            }
75
76            self.execute(clean_statement, &[]).await?;
77        }
78
79        Ok(())
80    }
81
82    async fn query_as<T: DinocoRow + Send + 'static>(
83        &self,
84        query: &str,
85        params: &[DinocoValue],
86    ) -> DinocoResult<Vec<T>> {
87        if let Ok(tx_conn) = SQLITE_TX_CONNECTION.try_with(Clone::clone) {
88            let conn = tx_conn.lock().await;
89
90            return query_as_with_connection::<T>(&conn, query, params, &self.query_logger).await;
91        }
92
93        let conn = self.pool.get().await.map_err(DinocoError::from)?;
94
95        query_as_with_connection::<T>(&conn, query, params, &self.query_logger).await
96    }
97}
98
99impl DinocoTransactionAdapter for SqliteAdapter {
100    fn with_transaction<'a, T, F>(&'a self, operation: F) -> Pin<Box<dyn Future<Output = DinocoResult<T>> + Send + 'a>>
101    where
102        T: Send + 'a,
103        F: FnOnce() -> Pin<Box<dyn Future<Output = DinocoResult<T>> + Send + 'a>> + Send + 'a,
104    {
105        Box::pin(async move {
106            if SQLITE_TX_CONNECTION.try_with(|_| ()).is_ok() {
107                return operation().await;
108            }
109
110            let connection = self.pool.get().await.map_err(DinocoError::from)?;
111            let tx_connection = Arc::new(tokio::sync::Mutex::new(connection));
112
113            {
114                let conn = tx_connection.lock().await;
115                conn.interact(|conn| conn.execute("BEGIN", []))
116                    .await
117                    .map_err(DinocoError::from)?
118                    .map_err(DinocoError::from)?;
119            }
120
121            let result = SQLITE_TX_CONNECTION.scope(tx_connection.clone(), async move { operation().await }).await;
122
123            match result {
124                Ok(output) => {
125                    let conn = tx_connection.lock().await;
126                    conn.interact(|conn| conn.execute("COMMIT", []))
127                        .await
128                        .map_err(DinocoError::from)?
129                        .map_err(DinocoError::from)?;
130
131                    Ok(output)
132                }
133                Err(error) => {
134                    let conn = tx_connection.lock().await;
135                    let _ = conn.interact(|conn| conn.execute("ROLLBACK", [])).await;
136
137                    Err(error)
138                }
139            }
140        })
141    }
142}
143
144async fn execute_result_with_connection(
145    conn: &deadpool_sqlite::Object,
146    query: &str,
147    params: &[DinocoValue],
148    query_logger: &DinocoQueryLogger,
149) -> DinocoResult<ExecutionResult> {
150    let query_owned = query.to_string();
151    let params_owned = params.to_vec();
152    let logged_query = query.to_string();
153    let logged_params = params.to_vec();
154    let started_at = Instant::now();
155
156    let affected_rows = conn
157        .interact(move |conn| {
158            let params_refs: Vec<&dyn rusqlite::ToSql> =
159                params_owned.iter().map(|p| p as &dyn rusqlite::ToSql).collect();
160
161            conn.execute(&query_owned, params_refs.as_slice())
162                .map(|affected_rows| (affected_rows, conn.last_insert_rowid()))
163        })
164        .await
165        .map_err(DinocoError::from)?
166        .map_err(DinocoError::from)?;
167
168    query_logger.log(DinocoQueryLog {
169        adapter: "sqlite",
170        duration: started_at.elapsed(),
171        params: logged_params,
172        query: logged_query,
173    });
174
175    Ok(ExecutionResult { affected_rows: affected_rows.0 as u64, last_insert_id: Some(affected_rows.1) })
176}
177
178async fn query_as_with_connection<T: DinocoRow + Send + 'static>(
179    conn: &deadpool_sqlite::Object,
180    query: &str,
181    params: &[DinocoValue],
182    query_logger: &DinocoQueryLogger,
183) -> DinocoResult<Vec<T>> {
184    let query_owned = query.to_string();
185    let params_owned = params.to_vec();
186    let logged_query = query.to_string();
187    let logged_params = params.to_vec();
188    let started_at = Instant::now();
189
190    let results = conn
191        .interact(move |conn| -> DinocoResult<Vec<T>> {
192            let mut stmt = conn.prepare(&query_owned).map_err(DinocoError::from)?;
193            let params_refs: Vec<&dyn rusqlite::ToSql> =
194                params_owned.iter().map(|p| p as &dyn rusqlite::ToSql).collect();
195
196            let mut rows = stmt.query(params_refs.as_slice()).map_err(DinocoError::from)?;
197            let mut results = Vec::new();
198
199            while let Some(row) = rows.next().map_err(DinocoError::from)? {
200                results.push(T::from_row(row)?);
201            }
202
203            Ok(results)
204        })
205        .await
206        .map_err(DinocoError::from)??;
207
208    query_logger.log(DinocoQueryLog {
209        adapter: "sqlite",
210        duration: started_at.elapsed(),
211        params: logged_params,
212        query: logged_query,
213    });
214
215    Ok(results)
216}
217
218impl rusqlite::ToSql for DinocoValue {
219    fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
220        match self {
221            DinocoValue::Null => Ok(ToSqlOutput::Owned(Value::Null)),
222            DinocoValue::Integer(i) => Ok(ToSqlOutput::Owned(Value::Integer(*i))),
223            DinocoValue::Float(f) => Ok(ToSqlOutput::Owned(Value::Real(*f))),
224            DinocoValue::Boolean(b) => Ok(ToSqlOutput::Owned(Value::Integer(if *b { 1 } else { 0 }))),
225            DinocoValue::String(s) => Ok(ToSqlOutput::Owned(Value::Text(s.clone()))),
226            DinocoValue::Enum(_, s) => Ok(ToSqlOutput::Owned(Value::Text(s.clone()))),
227            DinocoValue::Json(v) => Ok(ToSqlOutput::Owned(Value::Text(v.to_string()))),
228            DinocoValue::Bytes(v) => Ok(ToSqlOutput::Owned(Value::Blob(v.clone()))),
229            DinocoValue::DateTime(dt) => Ok(ToSqlOutput::Owned(Value::Text(dt.to_string()))),
230            DinocoValue::Date(date) => Ok(ToSqlOutput::Owned(Value::Text(date.to_string()))),
231        }
232    }
233}
234
235impl From<deadpool_sqlite::CreatePoolError> for DinocoError {
236    fn from(e: deadpool_sqlite::CreatePoolError) -> Self {
237        Self::ConnectionError(format!("Failed to get connection from pool: {}", e))
238    }
239}
240
241impl From<deadpool_sqlite::PoolError> for DinocoError {
242    fn from(e: deadpool_sqlite::PoolError) -> Self {
243        Self::ConnectionError(format!("Failed to get connection from pool: {}", e))
244    }
245}
246
247impl From<deadpool_sqlite::BuildError> for DinocoError {
248    fn from(e: deadpool_sqlite::BuildError) -> Self {
249        Self::ConnectionError(format!("Failed to build connection pool: {}", e))
250    }
251}
252
253impl From<deadpool_sqlite::InteractError> for DinocoError {
254    fn from(e: deadpool_sqlite::InteractError) -> Self {
255        Self::ParseError(e.to_string())
256    }
257}
258
259impl From<rusqlite::Error> for DinocoError {
260    fn from(e: rusqlite::Error) -> Self {
261        if let Some(error) = map_sqlite_constraint_error(&e) {
262            return Self::Constraint(error);
263        }
264
265        Self::Sqlite(e)
266    }
267}
268
269fn map_sqlite_constraint_error(error: &rusqlite::Error) -> Option<ConstraintError> {
270    let rusqlite::Error::SqliteFailure(_, message) = error else {
271        return None;
272    };
273    let message = message.clone()?;
274    let normalized = message.to_ascii_lowercase();
275
276    if normalized.starts_with("unique constraint failed:") {
277        let targets = parse_sqlite_constraint_targets(&message, "UNIQUE constraint failed:");
278        let table = targets.first().and_then(|target| target.split('.').next()).map(str::to_string);
279        let columns =
280            targets.into_iter().map(|target| target.split('.').nth(1).unwrap_or(target.as_str()).to_string()).collect();
281
282        return Some(ConstraintError::unique(table, columns, None, message));
283    }
284
285    if normalized.starts_with("not null constraint failed:") {
286        let targets = parse_sqlite_constraint_targets(&message, "NOT NULL constraint failed:");
287        let table = targets.first().and_then(|target| target.split('.').next()).map(str::to_string);
288        let columns =
289            targets.into_iter().map(|target| target.split('.').nth(1).unwrap_or(target.as_str()).to_string()).collect();
290
291        return Some(ConstraintError::not_null(table, columns, None, message));
292    }
293
294    if normalized.starts_with("foreign key constraint failed") {
295        return Some(ConstraintError::foreign_key(None, Vec::new(), None, message));
296    }
297
298    if normalized.starts_with("check constraint failed:") {
299        let constraint =
300            message.split_once(':').map(|(_, rest)| rest.trim().to_string()).filter(|item| !item.is_empty());
301
302        return Some(ConstraintError::check(None, Vec::new(), constraint, message));
303    }
304
305    None
306}
307
308fn parse_sqlite_constraint_targets(message: &str, prefix: &str) -> Vec<String> {
309    message
310        .strip_prefix(prefix)
311        .unwrap_or(message)
312        .split(',')
313        .map(str::trim)
314        .filter(|item| !item.is_empty())
315        .map(str::to_string)
316        .collect()
317}
dinoco_engine/databases/sqlite/mod.rs

dinoco_engine/databases/sqlite/
mod.rs
