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

sea_orm/driver/
sqlx_mysql.rs
