rust_ef_mysql/
connection.rs

1use async_trait::async_trait;
2use rust_ef::error::{EFError, EFResult};
3use rust_ef::provider::{DbValue, IAsyncConnection, IsolationLevel};
4use sqlx::Row;
5
6pub struct MySqlConnection {
7    conn: Option<sqlx::pool::PoolConnection<sqlx::MySql>>,
8    #[cfg(feature = "tracing")]
9    slow_query_threshold: Option<std::time::Duration>,
10}
11
12impl MySqlConnection {
13    pub(crate) fn new(conn: sqlx::pool::PoolConnection<sqlx::MySql>) -> Self {
14        Self {
15            conn: Some(conn),
16            #[cfg(feature = "tracing")]
17            slow_query_threshold: None,
18        }
19    }
20
21    fn threshold(&self) -> Option<std::time::Duration> {
22        #[cfg(feature = "tracing")]
23        {
24            self.slow_query_threshold
25        }
26        #[cfg(not(feature = "tracing"))]
27        {
28            None
29        }
30    }
31}
32
33#[async_trait]
34impl IAsyncConnection for MySqlConnection {
35    async fn execute(&mut self, sql: &str, params: &[DbValue]) -> EFResult<u64> {
36        let _guard = rust_ef::observability::QueryGuard::new(sql, self.threshold());
37        let conn = self
38            .conn
39            .as_mut()
40            .ok_or_else(|| EFError::connection("Connection already closed".to_string()))?;
41        let result = crate::type_conversion::build_mysql_query(sql, params)
42            .execute(&mut **conn)
43            .await
44            .map_err(|e| EFError::query(format!("Execution error: {}", e)))?;
45        Ok(result.rows_affected())
46    }
47
48    async fn query(&mut self, sql: &str, params: &[DbValue]) -> EFResult<Vec<Vec<DbValue>>> {
49        let _guard = rust_ef::observability::QueryGuard::new(sql, self.threshold());
50        let conn = self
51            .conn
52            .as_mut()
53            .ok_or_else(|| EFError::connection("Connection already closed".to_string()))?;
54        let rows = crate::type_conversion::build_mysql_query(sql, params)
55            .fetch_all(&mut **conn)
56            .await
57            .map_err(|e| EFError::query(format!("Query error: {}", e)))?;
58
59        if rows.is_empty() {
60            return Ok(Vec::new());
61        }
62
63        let result = rows
64            .iter()
65            .map(|row| {
66                row.columns()
67                    .iter()
68                    .enumerate()
69                    .map(|(i, _)| crate::row_conversion::cell_to_db_value(row, i))
70                    .collect()
71            })
72            .collect();
73
74        Ok(result)
75    }
76
77    async fn begin_transaction(&mut self) -> EFResult<()> {
78        self.execute("START TRANSACTION", &[]).await.map(|_| ())
79    }
80
81    async fn commit_transaction(&mut self) -> EFResult<()> {
82        self.execute("COMMIT", &[]).await.map(|_| ())
83    }
84
85    async fn rollback_transaction(&mut self) -> EFResult<()> {
86        self.execute("ROLLBACK", &[]).await.map(|_| ())
87    }
88
89    async fn create_savepoint(&mut self, name: &str) -> EFResult<()> {
90        self.execute(&format!("SAVEPOINT {}", name), &[])
91            .await
92            .map(|_| ())
93    }
94
95    async fn release_savepoint(&mut self, name: &str) -> EFResult<()> {
96        self.execute(&format!("RELEASE SAVEPOINT {}", name), &[])
97            .await
98            .map(|_| ())
99    }
100
101    async fn rollback_to_savepoint(&mut self, name: &str) -> EFResult<()> {
102        self.execute(&format!("ROLLBACK TO SAVEPOINT {}", name), &[])
103            .await
104            .map(|_| ())
105    }
106
107    async fn set_transaction_isolation(&mut self, level: IsolationLevel) -> EFResult<()> {
108        let sql = format!(
109            "SET TRANSACTION ISOLATION LEVEL {}",
110            match level {
111                IsolationLevel::ReadUncommitted => "READ UNCOMMITTED",
112                IsolationLevel::ReadCommitted => "READ COMMITTED",
113                IsolationLevel::RepeatableRead => "REPEATABLE READ",
114                IsolationLevel::Serializable => "SERIALIZABLE",
115            }
116        );
117        self.execute(&sql, &[]).await.map(|_| ())
118    }
119
120    #[cfg(feature = "tracing")]
121    fn set_slow_query_threshold(&mut self, threshold: std::time::Duration) {
122        self.slow_query_threshold = Some(threshold);
123    }
124}
rust_ef_mysql/connection.rs

rust_ef_mysql/
connection.rs
