Skip to main content

sz_orm_sqlx/
any_driver.rs

1//! Any driver — 一套代码多 DB 后端透明切换(SQLx 风格)
2//!
3//! SQLx 提供 `sqlx::Any` 适配器,让同一份代码可以在 MySQL/PostgreSQL/SQLite
4//! 之间透明切换。SZ-ORM 在 `sz-orm-sqlx` 已有各后端独立实现,
5//! 此模块在上层提供统一的 [`AnyConnection`] 和 [`AnyPool`] 抽象,
6//! 让运行时切换数据库后端成为可能。
7//!
8//! # 设计
9//!
10//! - [`AnyBackend`]:枚举后端类型
11//! - [`AnyPool`]:持有具体后端的 `Box<dyn ConnectionFactory>`
12//! - [`AnyConnection`]:持有具体后端的 `Box<dyn Connection>`
13//! - 通过 DSN 自动识别后端类型,运行时透明切换
14//!
15//! # 用法
16//!
17//! ```text
18//! use sz_orm_sqlx::any_driver::{AnyBackend, AnyPool};
19//!
20//! // 从 DSN 自动识别后端
21//! let pool = AnyPool::connect("mysql://root:pass@127.0.0.1/db").await?;
22//! let mut conn = pool.create().await?;
23//! let rows = conn.query("SELECT 1").await?;
24//!
25//! // 运行时切换后端
26//! let pg_pool = AnyPool::connect("postgres://user:pass@127.0.0.1/db").await?;
27//! let mut pg_conn = pg_pool.create().await?;
28//! let rows = pg_conn.query("SELECT 1").await?;
29//! ```
30
31use std::future::Future;
32use std::pin::Pin;
33use std::sync::Arc;
34use sz_orm_core::{
35    Connection, ConnectionFactory, DbError, Dialect, MySqlDialect, OracleDialect,
36    PostgreSqlDialect, QueryRows, QueryStreamItem, QueryValues, SqlServerDialect, SqliteDialect,
37    Value,
38};
39
40use sz_orm_core::DbType;
41
42use crate::any::{
43    MySqlPoolHandle, PgPoolHandle, SqlitePoolHandle, SqlxMySqlConnectionFactory,
44    SqlxPgConnectionFactory, SqlxSqliteConnectionFactory,
45};
46
47#[cfg(feature = "oracle")]
48use sz_orm_oracle::{OracleConnectionFactory, OraclePoolHandle};
49
50#[cfg(feature = "mssql")]
51use sz_orm_mssql::{MssqlConnectionFactory, MssqlPoolHandle};
52
53/// 数据库后端类型
54///
55/// v2.2.0 新增 `Oracle` 和 `Mssql` 变体(需启用 `oracle`/`mssql` feature)。
56/// `#[non_exhaustive]` 标注确保外部 crate match 时必须使用 wildcard 臂,
57/// 未来新增变体不会破坏现有代码。
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59#[non_exhaustive]
60pub enum AnyBackend {
61    /// MySQL / MariaDB
62    MySql,
63    /// PostgreSQL
64    Postgres,
65    /// SQLite
66    Sqlite,
67    /// Oracle(v2.2.0 新增,需启用 `oracle` feature)
68    Oracle,
69    /// SQL Server / MSSQL(v2.2.0 新增,需启用 `mssql` feature)
70    Mssql,
71}
72
73impl AnyBackend {
74    /// 从 DSN 自动识别后端类型
75    ///
76    /// # 支持的 scheme
77    ///
78    /// - `mysql://` / `mariadb://` → MySQL
79    /// - `postgres://` / `postgresql://` → Postgres
80    /// - `szrsql://` → Postgres(szrsql pgwire v3.0 协议,v10.13.2 新增)
81    /// - `sqlite://` / `sqlite:` → Sqlite
82    /// - `oracle://` → Oracle(v2.2.0 新增)
83    /// - `mssql://` / `sqlserver://` → Mssql(v2.2.0 新增)
84    pub fn from_dsn(dsn: &str) -> Result<Self, DbError> {
85        if dsn.starts_with("mysql://") || dsn.starts_with("mariadb://") {
86            Ok(AnyBackend::MySql)
87        } else if dsn.starts_with("postgres://") || dsn.starts_with("postgresql://") {
88            Ok(AnyBackend::Postgres)
89        } else if dsn.starts_with("szrsql://") {
90            // szrsql pgwire v3.0 走 PG 线协议:复用 sqlx Postgres 驱动,
91            // scheme 仅作识别标记,连接前改写回 postgres://
92            Ok(AnyBackend::Postgres)
93        } else if dsn.starts_with("sqlite://") || dsn.starts_with("sqlite:") {
94            Ok(AnyBackend::Sqlite)
95        } else if dsn.starts_with("oracle://") {
96            Ok(AnyBackend::Oracle)
97        } else if dsn.starts_with("mssql://") || dsn.starts_with("sqlserver://") {
98            Ok(AnyBackend::Mssql)
99        } else {
100            Err(DbError::ConnectionRefused(format!(
101                "未知的 DSN scheme: {}(支持 mysql/postgres/sqlite/oracle/mssql)",
102                dsn
103            )))
104        }
105    }
106
107    /// 从 [`DbType`] 转换为 [`AnyBackend`]
108    ///
109    /// 返回 `None` 的情况:DbType 对应的后端不在 sz-orm-sqlx 支持范围内
110    /// (如 Redis、MongoDB、ClickHouse 等非关系型数据库)。
111    ///
112    /// # 映射
113    ///
114    /// - `DbType::MySQL` / `DbType::MariaDB` / `DbType::TiDB` / `DbType::OceanBase` → MySql
115    /// - `DbType::PostgreSQL` / `DbType::Kingbase` / `DbType::PolarDB` / `DbType::GaussDB` → Postgres
116    /// - `DbType::Sqlite` → Sqlite
117    /// - `DbType::Oracle` / `DbType::Dameng` → Oracle
118    /// - `DbType::SqlServer` / `DbType::Sybase` / `DbType::GBase` → Mssql
119    /// - 其他 → None
120    pub fn from_db_type(db_type: DbType) -> Option<Self> {
121        match db_type {
122            DbType::MySQL | DbType::MariaDB | DbType::TiDB | DbType::OceanBase => {
123                Some(AnyBackend::MySql)
124            }
125            DbType::PostgreSQL
126            | DbType::Kingbase
127            | DbType::PolarDB
128            | DbType::GaussDB
129            | DbType::Szrsql => Some(AnyBackend::Postgres),
130            DbType::Sqlite => Some(AnyBackend::Sqlite),
131            DbType::Oracle | DbType::Dameng => Some(AnyBackend::Oracle),
132            DbType::SqlServer | DbType::Sybase | DbType::GBase => Some(AnyBackend::Mssql),
133            _ => None,
134        }
135    }
136
137    /// 后端名称
138    pub fn name(&self) -> &'static str {
139        match self {
140            AnyBackend::MySql => "mysql",
141            AnyBackend::Postgres => "postgres",
142            AnyBackend::Sqlite => "sqlite",
143            AnyBackend::Oracle => "oracle",
144            AnyBackend::Mssql => "mssql",
145        }
146    }
147
148    /// 返回对应后端的 Dialect 实例(v2.2.0 新增)
149    ///
150    /// - MySql → [`MySqlDialect`]
151    /// - Postgres → [`PostgreSqlDialect`]
152    /// - Sqlite → [`SqliteDialect`]
153    /// - Oracle → [`OracleDialect`]
154    /// - Mssql → [`SqlServerDialect`]
155    pub fn dialect(&self) -> Box<dyn Dialect> {
156        match self {
157            AnyBackend::MySql => Box::new(MySqlDialect),
158            AnyBackend::Postgres => Box::new(PostgreSqlDialect),
159            AnyBackend::Sqlite => Box::new(SqliteDialect),
160            AnyBackend::Oracle => Box::new(OracleDialect),
161            AnyBackend::Mssql => Box::new(SqlServerDialect),
162        }
163    }
164}
165
166/// 后端无关的连接工厂
167pub struct AnyPool {
168    backend: AnyBackend,
169    factory: Arc<dyn ConnectionFactory>,
170}
171
172impl AnyPool {
173    /// 连接数据库,根据 DSN 自动识别后端
174    ///
175    /// # 错误
176    ///
177    /// - DSN scheme 不识别 → [`DbError::ConnectionRefused`]
178    /// - 连接失败 → [`DbError::ConnectionError`]
179    /// - Oracle/MSSQL 后端未启用对应 feature → [`DbError::ConnectionRefused`] 含提示
180    pub async fn connect(dsn: &str) -> Result<Self, DbError> {
181        let backend = AnyBackend::from_dsn(dsn)?;
182        let factory: Arc<dyn ConnectionFactory> = match backend {
183            AnyBackend::MySql => {
184                let handle = Arc::new(MySqlPoolHandle::connect(dsn).await?);
185                Arc::new(SqlxMySqlConnectionFactory::new(handle))
186            }
187            AnyBackend::Postgres => {
188                let handle = Arc::new(PgPoolHandle::connect(dsn).await?);
189                Arc::new(SqlxPgConnectionFactory::new(handle))
190            }
191            AnyBackend::Sqlite => {
192                let handle = Arc::new(SqlitePoolHandle::connect(dsn).await?);
193                Arc::new(SqlxSqliteConnectionFactory::new(handle))
194            }
195            AnyBackend::Oracle => {
196                #[cfg(feature = "oracle")]
197                {
198                    let (username, password, connect_string) = parse_oracle_dsn(dsn)?;
199                    let handle = Arc::new(OraclePoolHandle::connect(
200                        &username,
201                        &password,
202                        &connect_string,
203                    )?);
204                    Arc::new(OracleConnectionFactory::new(handle))
205                }
206                #[cfg(not(feature = "oracle"))]
207                {
208                    return Err(DbError::ConnectionRefused(
209                        "Oracle 后端未启用,请在 Cargo.toml 中添加 features = [\"oracle\"]"
210                            .to_string(),
211                    ));
212                }
213            }
214            AnyBackend::Mssql => {
215                #[cfg(feature = "mssql")]
216                {
217                    let ado_string = parse_mssql_dsn(dsn)?;
218                    let handle = Arc::new(MssqlPoolHandle::connect(&ado_string).await?);
219                    Arc::new(MssqlConnectionFactory::new(handle))
220                }
221                #[cfg(not(feature = "mssql"))]
222                {
223                    return Err(DbError::ConnectionRefused(
224                        "MSSQL 后端未启用,请在 Cargo.toml 中添加 features = [\"mssql\"]"
225                            .to_string(),
226                    ));
227                }
228            }
229        };
230        Ok(Self { backend, factory })
231    }
232
233    /// 从已有的连接工厂构造
234    pub fn from_factory(backend: AnyBackend, factory: Arc<dyn ConnectionFactory>) -> Self {
235        Self { backend, factory }
236    }
237
238    /// 获取后端类型
239    pub fn backend(&self) -> AnyBackend {
240        self.backend
241    }
242
243    /// 返回对应后端的 Dialect 实例(v2.2.0 新增)
244    ///
245    /// 委托 [`AnyBackend::dialect()`],根据后端自动选择方言。
246    pub fn dialect(&self) -> Box<dyn Dialect> {
247        self.backend.dialect()
248    }
249
250    /// 创建一个新连接
251    pub async fn create(&self) -> Result<AnyConnection, DbError> {
252        let conn = self.factory.create().await?;
253        Ok(AnyConnection {
254            backend: self.backend,
255            inner: conn,
256        })
257    }
258}
259
260/// 统一连接便利函数:按 DSN scheme 自动识别后端并创建连接
261///
262/// 这是 sz-orm-sqlx 对外提供的统一入口,下游项目(如 sz-rust)只需调用此函数,
263/// 无需为每种数据库编写独立的连接逻辑。Oracle/MSSQL 支持通过 feature gate 启用。
264///
265/// # 支持的 DSN scheme
266///
267/// - `mysql://` / `mariadb://` → MySQL
268/// - `postgres://` / `postgresql://` → PostgreSQL
269/// - `sqlite://` / `sqlite:` → SQLite
270/// - `oracle://` → Oracle(需启用 `oracle` feature)
271/// - `mssql://` / `sqlserver://` → MSSQL(需启用 `mssql` feature)
272///
273/// # 错误
274///
275/// - DSN scheme 不识别 → [`DbError::ConnectionRefused`]
276/// - Oracle/MSSQL 后端未启用对应 feature → [`DbError::ConnectionRefused`] 含提示
277/// - 连接失败 → [`DbError::ConnectionError`]
278///
279/// # 示例
280///
281/// ```no_run
282/// # #[tokio::main]
283/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
284/// use sz_orm_core::Connection;
285/// let mut conn = sz_orm_sqlx::create_connection("sqlite::memory:").await?;
286/// conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)").await?;
287/// # Ok(())
288/// # }
289/// ```
290pub async fn create_connection(dsn: &str) -> Result<Box<dyn Connection>, DbError> {
291    let pool = AnyPool::connect(dsn).await?;
292    let conn = pool.create().await?;
293    Ok(Box::new(conn))
294}
295
296/// 按 [`DbType`] 创建连接
297///
298/// 当 DSN 的 scheme 可能与 `db_type` 不一致时使用此函数。
299/// 会根据 `db_type` 修正 DSN 的 scheme 后再连接。
300///
301/// # 示例
302///
303/// ```no_run
304/// # #[tokio::main]
305/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
306/// use sz_orm_core::{Connection, DbType};
307/// let mut conn = sz_orm_sqlx::create_connection_by_type(
308///     DbType::Sqlite, "sqlite::memory:"
309/// ).await?;
310/// conn.execute("SELECT 1").await?;
311/// # Ok(())
312/// # }
313/// ```
314pub async fn create_connection_by_type(
315    db_type: DbType,
316    dsn: &str,
317) -> Result<Box<dyn Connection>, DbError> {
318    let backend = AnyBackend::from_db_type(db_type).ok_or_else(|| {
319        DbError::ConnectionRefused(format!(
320            "DbType {:?} 不在 sz-orm-sqlx 支持范围内(支持 MySQL/PostgreSQL/SQLite/Oracle/MSSQL 及兼容方言)",
321            db_type
322        ))
323    })?;
324    let normalized = normalize_dsn_scheme(backend, dsn);
325    create_connection(&normalized).await
326}
327
328/// 根据 [`AnyBackend`] 修正 DSN 的 scheme
329///
330/// 如果 DSN 已有正确的 scheme 则原样返回;
331/// 否则替换 scheme 部分以匹配目标后端。
332fn normalize_dsn_scheme(backend: AnyBackend, dsn: &str) -> String {
333    let expected_scheme = match backend {
334        AnyBackend::MySql => "mysql://",
335        AnyBackend::Postgres => "postgres://",
336        AnyBackend::Sqlite => "sqlite:",
337        AnyBackend::Oracle => "oracle://",
338        AnyBackend::Mssql => "mssql://",
339    };
340    let known_schemes = [
341        "mysql://",
342        "mariadb://",
343        "postgres://",
344        "postgresql://",
345        "szrsql://",
346        "sqlite://",
347        "sqlite:",
348        "oracle://",
349        "mssql://",
350        "sqlserver://",
351    ];
352    for scheme in &known_schemes {
353        if let Some(rest) = dsn.strip_prefix(scheme) {
354            if backend == AnyBackend::from_dsn(dsn).unwrap_or(AnyBackend::MySql) {
355                return dsn.to_string();
356            }
357            return format!("{}{}", expected_scheme, rest);
358        }
359    }
360    dsn.to_string()
361}
362
363/// 解析 Oracle DSN 为 (username, password, connect_string)
364///
365/// 格式:`oracle://user:pass@host:port/service`
366///
367/// 返回:username="user", password="pass", connect_string="host:port/service"
368#[allow(dead_code)]
369pub(crate) fn parse_oracle_dsn(dsn: &str) -> Result<(String, String, String), DbError> {
370    let rest = dsn
371        .strip_prefix("oracle://")
372        .ok_or_else(|| DbError::ConnectionRefused(format!("无效的 Oracle DSN: {}", dsn)))?;
373    parse_user_pass_host(rest, "Oracle")
374}
375
376/// 解析 MSSQL DSN 为 ADO.NET 连接字符串
377///
378/// 格式:`mssql://user:pass@host:port/database` 或 `sqlserver://user:pass@host:port/database`
379///
380/// 返回:`Server=host,port;Database=database;User Id=user;Password=pass;`
381#[allow(dead_code)]
382pub(crate) fn parse_mssql_dsn(dsn: &str) -> Result<String, DbError> {
383    let rest = dsn
384        .strip_prefix("mssql://")
385        .or_else(|| dsn.strip_prefix("sqlserver://"))
386        .ok_or_else(|| DbError::ConnectionRefused(format!("无效的 MSSQL DSN: {}", dsn)))?;
387    let (username, password, host_port_db) = parse_user_pass_host(rest, "MSSQL")?;
388    let (host_port, database) = host_port_db
389        .split_once('/')
390        .ok_or_else(|| DbError::ConnectionRefused(format!("MSSQL DSN 缺少 database: {}", dsn)))?;
391    let (host, port) = host_port.split_once(':').unwrap_or((host_port, "1433"));
392    Ok(format!(
393        "Server={},{};Database={};User Id={};Password={};",
394        host, port, database, username, password
395    ))
396}
397
398/// 通用 DSN 解析:`user:pass@host:port/...` → (user, pass, host:port/...)
399#[allow(dead_code)]
400fn parse_user_pass_host(rest: &str, backend: &str) -> Result<(String, String, String), DbError> {
401    let (userinfo, hostinfo) = rest
402        .split_once('@')
403        .ok_or_else(|| DbError::ConnectionRefused(format!("{} DSN 缺少 @: {}", backend, rest)))?;
404    let (username, password) = userinfo.split_once(':').ok_or_else(|| {
405        DbError::ConnectionRefused(format!("{} DSN 缺少 password: {}", backend, rest))
406    })?;
407    Ok((
408        username.to_string(),
409        password.to_string(),
410        hostinfo.to_string(),
411    ))
412}
413
414/// 后端无关的连接
415pub struct AnyConnection {
416    backend: AnyBackend,
417    inner: Box<dyn Connection>,
418}
419
420impl AnyConnection {
421    /// 获取后端类型
422    pub fn backend(&self) -> AnyBackend {
423        self.backend
424    }
425}
426
427impl Connection for AnyConnection {
428    fn db_type(&self) -> Option<DbType> {
429        Some(match self.backend {
430            AnyBackend::MySql => DbType::MySQL,
431            AnyBackend::Postgres => DbType::PostgreSQL,
432            AnyBackend::Sqlite => DbType::Sqlite,
433            AnyBackend::Oracle => DbType::Oracle,
434            AnyBackend::Mssql => DbType::SqlServer,
435        })
436    }
437
438    fn execute<'a>(
439        &'a mut self,
440        sql: &'a str,
441    ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
442        self.inner.execute(sql)
443    }
444
445    fn query<'a>(
446        &'a mut self,
447        sql: &'a str,
448    ) -> Pin<Box<dyn Future<Output = Result<QueryRows, DbError>> + Send + 'a>> {
449        self.inner.query(sql)
450    }
451
452    fn begin_transaction<'a>(
453        &'a mut self,
454    ) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
455        self.inner.begin_transaction()
456    }
457
458    fn commit<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
459        self.inner.commit()
460    }
461
462    fn rollback<'a>(
463        &'a mut self,
464    ) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
465        self.inner.rollback()
466    }
467
468    fn in_transaction(&self) -> bool {
469        self.inner.in_transaction()
470    }
471
472    fn session_reset_sql(&self) -> Option<&'static str> {
473        self.inner.session_reset_sql()
474    }
475
476    fn is_connected(&self) -> bool {
477        self.inner.is_connected()
478    }
479
480    fn ping<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>> {
481        self.inner.ping()
482    }
483
484    fn close<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
485        self.inner.close()
486    }
487
488    fn execute_with_params<'a>(
489        &'a mut self,
490        sql: &'a str,
491        params: &'a [Value],
492    ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
493        self.inner.execute_with_params(sql, params)
494    }
495
496    fn query_with_params<'a>(
497        &'a mut self,
498        sql: &'a str,
499        params: &'a [Value],
500    ) -> Pin<Box<dyn Future<Output = Result<QueryRows, DbError>> + Send + 'a>> {
501        self.inner.query_with_params(sql, params)
502    }
503
504    fn query_values<'a>(
505        &'a mut self,
506        sql: &'a str,
507    ) -> Pin<Box<dyn Future<Output = Result<QueryValues, DbError>> + Send + 'a>> {
508        self.inner.query_values(sql)
509    }
510
511    fn query_values_with_params<'a>(
512        &'a mut self,
513        sql: &'a str,
514        params: &'a [Value],
515    ) -> Pin<Box<dyn Future<Output = Result<QueryValues, DbError>> + Send + 'a>> {
516        self.inner.query_values_with_params(sql, params)
517    }
518
519    fn query_stream<'a>(
520        &'a mut self,
521        sql: &'a str,
522    ) -> Pin<Box<dyn futures::Stream<Item = QueryStreamItem> + Send + 'a>> {
523        self.inner.query_stream(sql)
524    }
525
526    fn query_stream_cursor<'a>(
527        &'a mut self,
528        sql: &'a str,
529        batch_size: usize,
530    ) -> Pin<Box<dyn futures::Stream<Item = QueryStreamItem> + Send + 'a>> {
531        self.inner.query_stream_cursor(sql, batch_size)
532    }
533
534    fn execute_batch<'a>(
535        &'a mut self,
536        sqls: &'a [String],
537    ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
538        self.inner.execute_batch(sqls)
539    }
540}
541
542#[cfg(test)]
543mod tests {
544    use super::*;
545
546    #[test]
547    fn test_any_backend_from_dsn_mysql() {
548        assert_eq!(
549            AnyBackend::from_dsn("mysql://root:pass@127.0.0.1/db").unwrap(),
550            AnyBackend::MySql
551        );
552        assert_eq!(
553            AnyBackend::from_dsn("mariadb://root:pass@127.0.0.1/db").unwrap(),
554            AnyBackend::MySql
555        );
556    }
557
558    #[test]
559    fn test_any_backend_from_dsn_postgres() {
560        assert_eq!(
561            AnyBackend::from_dsn("postgres://user:pass@127.0.0.1/db").unwrap(),
562            AnyBackend::Postgres
563        );
564        assert_eq!(
565            AnyBackend::from_dsn("postgresql://user:pass@127.0.0.1/db").unwrap(),
566            AnyBackend::Postgres
567        );
568    }
569
570    #[test]
571    fn test_any_backend_from_dsn_sqlite() {
572        assert_eq!(
573            AnyBackend::from_dsn("sqlite::memory:").unwrap(),
574            AnyBackend::Sqlite
575        );
576        assert_eq!(
577            AnyBackend::from_dsn("sqlite://./test.db").unwrap(),
578            AnyBackend::Sqlite
579        );
580    }
581
582    #[test]
583    fn test_any_backend_from_dsn_oracle() {
584        assert_eq!(
585            AnyBackend::from_dsn("oracle://sys:test123@127.0.0.1:1521/freepdb1").unwrap(),
586            AnyBackend::Oracle
587        );
588    }
589
590    #[test]
591    fn test_any_backend_from_dsn_mssql() {
592        assert_eq!(
593            AnyBackend::from_dsn("mssql://sa:test123@127.0.0.1:1433/testdb").unwrap(),
594            AnyBackend::Mssql
595        );
596    }
597
598    #[test]
599    fn test_any_backend_from_dsn_sqlserver() {
600        assert_eq!(
601            AnyBackend::from_dsn("sqlserver://sa:test123@127.0.0.1:1433/testdb").unwrap(),
602            AnyBackend::Mssql
603        );
604    }
605
606    #[test]
607    fn test_any_backend_from_dsn_unknown_v22() {
608        let result = AnyBackend::from_dsn("redis://127.0.0.1");
609        assert!(result.is_err());
610        if let Err(e) = result {
611            let msg = format!("{}", e);
612            assert!(msg.contains("mysql"));
613            assert!(msg.contains("postgres"));
614            assert!(msg.contains("sqlite"));
615            assert!(msg.contains("oracle"));
616            assert!(msg.contains("mssql"));
617        }
618    }
619
620    #[test]
621    fn test_any_backend_name_v22() {
622        assert_eq!(AnyBackend::MySql.name(), "mysql");
623        assert_eq!(AnyBackend::Postgres.name(), "postgres");
624        assert_eq!(AnyBackend::Sqlite.name(), "sqlite");
625        assert_eq!(AnyBackend::Oracle.name(), "oracle");
626        assert_eq!(AnyBackend::Mssql.name(), "mssql");
627    }
628
629    #[test]
630    fn test_any_backend_equality() {
631        assert_eq!(AnyBackend::MySql, AnyBackend::MySql);
632        assert_ne!(AnyBackend::MySql, AnyBackend::Postgres);
633        assert_ne!(AnyBackend::Postgres, AnyBackend::Sqlite);
634        assert_ne!(AnyBackend::Oracle, AnyBackend::Mssql);
635        assert_ne!(AnyBackend::Oracle, AnyBackend::MySql);
636    }
637
638    #[test]
639    fn test_parse_oracle_dsn() {
640        let (user, pass, cs) =
641            parse_oracle_dsn("oracle://sys:test123@127.0.0.1:1521/freepdb1").unwrap();
642        assert_eq!(user, "sys");
643        assert_eq!(pass, "test123");
644        assert_eq!(cs, "127.0.0.1:1521/freepdb1");
645    }
646
647    #[test]
648    fn test_parse_mssql_dsn() {
649        let ado = parse_mssql_dsn("mssql://sa:test123@127.0.0.1:1433/testdb").unwrap();
650        assert!(ado.contains("Server=127.0.0.1,1433"));
651        assert!(ado.contains("Database=testdb"));
652        assert!(ado.contains("User Id=sa"));
653        assert!(ado.contains("Password=test123"));
654    }
655
656    #[test]
657    fn test_parse_mssql_dsn_sqlserver_scheme() {
658        let ado = parse_mssql_dsn("sqlserver://sa:pass@localhost/testdb").unwrap();
659        assert!(ado.contains("Server=localhost,1433"));
660        assert!(ado.contains("Database=testdb"));
661    }
662
663    // ---- 真实 SQLite 集成测试 ----
664
665    #[tokio::test]
666    async fn test_any_pool_sqlite_connect_and_query() {
667        let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
668        assert_eq!(pool.backend(), AnyBackend::Sqlite);
669
670        let mut conn = pool.create().await.unwrap();
671        assert_eq!(conn.backend(), AnyBackend::Sqlite);
672
673        // 创建表并插入数据
674        conn.execute("CREATE TABLE test_any (id INTEGER PRIMARY KEY, name TEXT)")
675            .await
676            .unwrap();
677        conn.execute("INSERT INTO test_any (name) VALUES ('Alice')")
678            .await
679            .unwrap();
680        conn.execute("INSERT INTO test_any (name) VALUES ('Bob')")
681            .await
682            .unwrap();
683
684        // 查询验证
685        let rows = conn
686            .query("SELECT * FROM test_any ORDER BY id")
687            .await
688            .unwrap();
689        assert_eq!(rows.len(), 2);
690        assert_eq!(rows[0].get("name").and_then(|v| v.as_str()), Some("Alice"));
691        assert_eq!(rows[1].get("name").and_then(|v| v.as_str()), Some("Bob"));
692    }
693
694    #[tokio::test]
695    async fn test_any_pool_sqlite_transaction_commit() {
696        let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
697        let mut conn = pool.create().await.unwrap();
698
699        conn.execute("CREATE TABLE tx_test (id INTEGER PRIMARY KEY, val INTEGER)")
700            .await
701            .unwrap();
702
703        // 事务提交
704        conn.begin_transaction().await.unwrap();
705        conn.execute("INSERT INTO tx_test (val) VALUES (1)")
706            .await
707            .unwrap();
708        conn.execute("INSERT INTO tx_test (val) VALUES (2)")
709            .await
710            .unwrap();
711        conn.commit().await.unwrap();
712
713        let rows = conn.query("SELECT * FROM tx_test").await.unwrap();
714        assert_eq!(rows.len(), 2);
715    }
716
717    #[tokio::test]
718    async fn test_any_pool_sqlite_transaction_rollback() {
719        let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
720        let mut conn = pool.create().await.unwrap();
721
722        conn.execute("CREATE TABLE tx_rb (id INTEGER PRIMARY KEY, val INTEGER)")
723            .await
724            .unwrap();
725
726        conn.begin_transaction().await.unwrap();
727        conn.execute("INSERT INTO tx_rb (val) VALUES (1)")
728            .await
729            .unwrap();
730        conn.rollback().await.unwrap();
731
732        let rows = conn.query("SELECT * FROM tx_rb").await.unwrap();
733        assert_eq!(rows.len(), 0);
734    }
735
736    #[tokio::test]
737    async fn test_any_connection_in_transaction_state_tracking() {
738        let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
739        let mut conn = pool.create().await.unwrap();
740
741        assert!(!conn.in_transaction(), "新建连接不应处于事务中");
742
743        conn.begin_transaction().await.unwrap();
744        assert!(conn.in_transaction(), "begin 后应处于事务中");
745
746        conn.execute("CREATE TABLE tx_track (id INTEGER PRIMARY KEY)")
747            .await
748            .unwrap();
749        assert!(conn.in_transaction(), "事务内执行 SQL 后仍应处于事务中");
750
751        conn.commit().await.unwrap();
752        assert!(!conn.in_transaction(), "commit 后不应处于事务中");
753
754        conn.begin_transaction().await.unwrap();
755        assert!(conn.in_transaction());
756        conn.rollback().await.unwrap();
757        assert!(!conn.in_transaction(), "rollback 后不应处于事务中");
758    }
759
760    #[tokio::test]
761    async fn test_any_connection_in_transaction_leak_detection() {
762        let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
763        let mut conn = pool.create().await.unwrap();
764
765        conn.execute("CREATE TABLE tx_leak (id INTEGER PRIMARY KEY)")
766            .await
767            .unwrap();
768
769        conn.begin_transaction().await.unwrap();
770        conn.execute("INSERT INTO tx_leak (id) VALUES (1)")
771            .await
772            .unwrap();
773
774        assert!(
775            conn.in_transaction(),
776            "未提交事务应被检测到(泄漏防护前提)"
777        );
778
779        conn.rollback().await.unwrap();
780        assert!(!conn.in_transaction());
781    }
782
783    #[tokio::test]
784    async fn test_any_pool_sqlite_ping() {
785        let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
786        let mut conn = pool.create().await.unwrap();
787        let ok = conn.ping().await;
788        assert!(ok);
789        assert!(conn.is_connected());
790    }
791
792    #[tokio::test]
793    async fn test_any_pool_invalid_dsn() {
794        let result = AnyPool::connect("invalid://dsn").await;
795        assert!(result.is_err());
796    }
797
798    #[tokio::test]
799    async fn test_any_pool_sqlite_count_query() {
800        let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
801        let mut conn = pool.create().await.unwrap();
802
803        conn.execute("CREATE TABLE cnt (id INTEGER PRIMARY KEY)")
804            .await
805            .unwrap();
806        for i in 1..=5 {
807            conn.execute(&format!("INSERT INTO cnt (id) VALUES ({})", i))
808                .await
809                .unwrap();
810        }
811
812        // 通过 SELECT * 验证行数(避开 sqlx 适配器中 COUNT(*) 类型推断的既有问题)
813        let rows = conn.query("SELECT * FROM cnt").await.unwrap();
814        assert_eq!(rows.len(), 5);
815    }
816
817    // ---- v2.2.0 A-2: Dialect 与 AnyPool 集成测试 ----
818
819    #[test]
820    fn test_any_backend_dialect_mapping() {
821        use sz_orm_core::DbType;
822
823        let mysql_d = AnyBackend::MySql.dialect();
824        assert_eq!(mysql_d.db_type(), DbType::MySQL);
825
826        let pg_d = AnyBackend::Postgres.dialect();
827        assert_eq!(pg_d.db_type(), DbType::PostgreSQL);
828
829        let sqlite_d = AnyBackend::Sqlite.dialect();
830        assert_eq!(sqlite_d.db_type(), DbType::Sqlite);
831
832        let oracle_d = AnyBackend::Oracle.dialect();
833        assert_eq!(oracle_d.db_type(), DbType::Oracle);
834
835        let mssql_d = AnyBackend::Mssql.dialect();
836        assert_eq!(mssql_d.db_type(), DbType::SqlServer);
837    }
838
839    #[test]
840    fn test_oracle_dialect_pagination() {
841        let d = AnyBackend::Oracle.dialect();
842        let sql = d.build_pagination("SELECT * FROM users", 2, 10);
843        let upper = sql.to_uppercase();
844        assert!(
845            upper.contains("OFFSET") || upper.contains("FETCH") || upper.contains("ROWNUM"),
846            "Oracle 分页 SQL 应含 OFFSET/FETCH/ROWNUM,实际: {}",
847            sql
848        );
849        assert!(
850            !upper.contains("LIMIT"),
851            "Oracle 分页 SQL 不应含 LIMIT,实际: {}",
852            sql
853        );
854    }
855
856    #[test]
857    fn test_mssql_dialect_pagination() {
858        let d = AnyBackend::Mssql.dialect();
859        let sql = d.build_pagination("SELECT * FROM users", 2, 10);
860        let upper = sql.to_uppercase();
861        assert!(
862            upper.contains("OFFSET") || upper.contains("FETCH"),
863            "MSSQL 分页 SQL 应含 OFFSET/FETCH,实际: {}",
864            sql
865        );
866        assert!(
867            !upper.contains("LIMIT"),
868            "MSSQL 分页 SQL 不应含 LIMIT,实际: {}",
869            sql
870        );
871    }
872
873    #[test]
874    fn test_oracle_dialect_no_placeholder() {
875        let d = AnyBackend::Oracle.dialect();
876        // 调用各 Dialect trait 方法,确认不 panic(无 todo!/unimplemented!)
877        let _ = d.db_type();
878        let _ = d.quote("col");
879        let _ = d.quote_checked("col").unwrap();
880        let _ = d.escape_string("val");
881        let _ = d.supports_returning();
882        let _ = d.build_pagination("SELECT 1", 1, 10);
883        let _ = d.json_type();
884        let _ = d.json_extract("col", "$.key");
885        let _ = d.full_text_search(&["col"], "kw");
886        let _ = d.bool_to_int("expr");
887        let _ = d.concat(&["a", "b"]);
888    }
889
890    #[test]
891    fn test_mssql_dialect_no_placeholder() {
892        let d = AnyBackend::Mssql.dialect();
893        let _ = d.db_type();
894        let _ = d.quote("col");
895        let _ = d.quote_checked("col").unwrap();
896        let _ = d.escape_string("val");
897        let _ = d.supports_returning();
898        let _ = d.build_pagination("SELECT 1", 1, 10);
899        let _ = d.json_type();
900        let _ = d.json_extract("col", "$.key");
901        let _ = d.full_text_search(&["col"], "kw");
902        let _ = d.bool_to_int("expr");
903        let _ = d.concat(&["a", "b"]);
904    }
905
906    #[tokio::test]
907    async fn test_any_pool_dialect_sqlite() {
908        let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
909        let d = pool.dialect();
910        assert_eq!(d.db_type(), sz_orm_core::DbType::Sqlite);
911    }
912
913    // ---- v6.9.0: from_db_type / create_connection / create_connection_by_type ----
914
915    #[test]
916    fn test_from_db_type_supported() {
917        use sz_orm_core::DbType;
918        assert_eq!(
919            AnyBackend::from_db_type(DbType::MySQL),
920            Some(AnyBackend::MySql)
921        );
922        assert_eq!(
923            AnyBackend::from_db_type(DbType::MariaDB),
924            Some(AnyBackend::MySql)
925        );
926        assert_eq!(
927            AnyBackend::from_db_type(DbType::TiDB),
928            Some(AnyBackend::MySql)
929        );
930        assert_eq!(
931            AnyBackend::from_db_type(DbType::OceanBase),
932            Some(AnyBackend::MySql)
933        );
934        assert_eq!(
935            AnyBackend::from_db_type(DbType::PostgreSQL),
936            Some(AnyBackend::Postgres)
937        );
938        assert_eq!(
939            AnyBackend::from_db_type(DbType::Kingbase),
940            Some(AnyBackend::Postgres)
941        );
942        assert_eq!(
943            AnyBackend::from_db_type(DbType::PolarDB),
944            Some(AnyBackend::Postgres)
945        );
946        assert_eq!(
947            AnyBackend::from_db_type(DbType::GaussDB),
948            Some(AnyBackend::Postgres)
949        );
950        assert_eq!(
951            AnyBackend::from_db_type(DbType::Sqlite),
952            Some(AnyBackend::Sqlite)
953        );
954        assert_eq!(
955            AnyBackend::from_db_type(DbType::Oracle),
956            Some(AnyBackend::Oracle)
957        );
958        assert_eq!(
959            AnyBackend::from_db_type(DbType::Dameng),
960            Some(AnyBackend::Oracle)
961        );
962        assert_eq!(
963            AnyBackend::from_db_type(DbType::SqlServer),
964            Some(AnyBackend::Mssql)
965        );
966        assert_eq!(
967            AnyBackend::from_db_type(DbType::Sybase),
968            Some(AnyBackend::Mssql)
969        );
970        assert_eq!(
971            AnyBackend::from_db_type(DbType::GBase),
972            Some(AnyBackend::Mssql)
973        );
974    }
975
976    #[test]
977    fn test_from_db_type_unsupported() {
978        use sz_orm_core::DbType;
979        assert_eq!(AnyBackend::from_db_type(DbType::Redis), None);
980        assert_eq!(AnyBackend::from_db_type(DbType::MongoDB), None);
981        assert_eq!(AnyBackend::from_db_type(DbType::ClickHouse), None);
982        assert_eq!(AnyBackend::from_db_type(DbType::VectorDb), None);
983    }
984
985    #[tokio::test]
986    async fn test_create_connection_sqlite() {
987        let mut conn = create_connection("sqlite::memory:").await.unwrap();
988        conn.execute("CREATE TABLE cc (id INTEGER PRIMARY KEY, v TEXT)")
989            .await
990            .unwrap();
991        conn.execute("INSERT INTO cc (v) VALUES ('ok')")
992            .await
993            .unwrap();
994        let rows = conn.query("SELECT * FROM cc").await.unwrap();
995        assert_eq!(rows.len(), 1);
996    }
997
998    #[tokio::test]
999    async fn test_create_connection_invalid_dsn() {
1000        let result = create_connection("invalid://dsn").await;
1001        assert!(result.is_err());
1002    }
1003
1004    #[tokio::test]
1005    async fn test_create_connection_by_type_sqlite() {
1006        use sz_orm_core::DbType;
1007        let mut conn = create_connection_by_type(DbType::Sqlite, "sqlite::memory:")
1008            .await
1009            .unwrap();
1010        conn.execute("SELECT 1").await.unwrap();
1011    }
1012
1013    #[tokio::test]
1014    async fn test_create_connection_by_type_unsupported() {
1015        use sz_orm_core::DbType;
1016        let result = create_connection_by_type(DbType::Redis, "redis://127.0.0.1").await;
1017        assert!(result.is_err());
1018    }
1019
1020    #[test]
1021    fn test_normalize_dsn_scheme_already_correct() {
1022        assert_eq!(
1023            normalize_dsn_scheme(AnyBackend::MySql, "mysql://root:pass@host/db"),
1024            "mysql://root:pass@host/db"
1025        );
1026        assert_eq!(
1027            normalize_dsn_scheme(AnyBackend::Postgres, "postgres://user@host/db"),
1028            "postgres://user@host/db"
1029        );
1030        assert_eq!(
1031            normalize_dsn_scheme(AnyBackend::Oracle, "oracle://sys:pass@host/svc"),
1032            "oracle://sys:pass@host/svc"
1033        );
1034    }
1035
1036    #[test]
1037    fn test_normalize_dsn_scheme_fix_mismatch() {
1038        let fixed = normalize_dsn_scheme(AnyBackend::MySql, "postgres://root:pass@host/db");
1039        assert!(fixed.starts_with("mysql://"));
1040        assert!(fixed.contains("root:pass@host/db"));
1041    }
1042}