Use generic connection trait

This should allow SeaORM to perform schema discovery with a DatabaseConnection

Also queries will not be run in parallel

Author: tyt2y3

Cargo.toml

@@ -34,7 +34,7 @@ name = "sea_schema"
 path = "src/lib.rs"
 
 [dependencies]
-futures = { version = "0.3", default-features = false, optional = true, features = ["alloc"] }
+async-trait = { version = "0.1", default-features = false }
 sea-schema-derive = { version = "0.3.0", path = "sea-schema-derive", default-features = false }
 sea-query = { version = "1.0.0-rc.14", default-features = false, features = ["derive"] }
 sea-query-sqlx = { version = "0.8.0-rc.1", default-features = false, optional = true }
@@ -50,7 +50,7 @@ postgres = ["sea-query/backend-postgres"]
 postgres-vector = ["sea-query/postgres-vector", "sea-query-sqlx/postgres-vector"]
 sqlite = ["sea-query/backend-sqlite"]
 def = []
-discovery = ["futures", "parser"]
+discovery = ["parser"]
 parser = ["query"]
 query = ["def"]
 writer = ["def"]
@@ -60,21 +60,18 @@ sqlx-dep = ["sqlx"]
 sqlx-all = ["sqlx-mysql", "sqlx-postgres", "sqlx-sqlite"]
 sqlx-mysql = [
     "mysql",
-    "futures",
     "sqlx-dep",
     "sea-query-sqlx/sqlx-mysql",
     "sqlx/mysql",
 ]
 sqlx-postgres = [
     "postgres",
-    "futures",
     "sqlx-dep",
     "sea-query-sqlx/sqlx-postgres",
     "sqlx/postgres",
 ]
 sqlx-sqlite = [
     "sqlite",
-    "futures",
     "sqlx-dep",
     "sea-query-sqlx/sqlx-sqlite",
     "sqlx/sqlite",

src/connection.rs

@@ -0,0 +1,9 @@
+use crate::sqlx_types::{SqlxError, SqlxRow};
+use sea_query::SelectStatement;
+
+#[async_trait::async_trait]
+pub trait Connection: Sized + Sync {
+    async fn query_all(&self, select: SelectStatement) -> Result<Vec<SqlxRow>, SqlxError>;
+
+    async fn query_all_raw(&self, sql: String) -> Result<Vec<SqlxRow>, SqlxError>;
+}

src/lib.rs

@@ -21,5 +21,8 @@ pub(crate) mod util;
 pub mod name;
 pub use name::*;
 
+mod connection;
+pub use connection::*;
+
 #[cfg(feature = "probe")]
 pub mod probe;

src/mysql/discovery/executor/mock.rs

@@ -1,28 +1,36 @@
-use crate::sqlx_types::{MySqlPool, mysql::MySqlRow};
+use crate::{
+    Connection,
+    sqlx_types::{MySqlPool, SqlxRow},
+};
 use sea_query::{MysqlQueryBuilder, SelectStatement};
 
 use crate::{debug_print, sqlx_types::SqlxError};
 
 #[allow(dead_code)]
-pub struct Executor {
-    pool: MySqlPool,
-}
+pub struct Executor {}
 
 pub trait IntoExecutor {
     fn into_executor(self) -> Executor;
 }
 
 impl IntoExecutor for MySqlPool {
     fn into_executor(self) -> Executor {
-        Executor { pool: self }
+        Executor {}
     }
 }
 
-impl Executor {
-    pub async fn fetch_all(&self, select: SelectStatement) -> Result<Vec<MySqlRow>, SqlxError> {
+#[async_trait::async_trait]
+impl Connection for Executor {
+    async fn query_all(&self, select: SelectStatement) -> Result<Vec<SqlxRow>, SqlxError> {
         let (_sql, _values) = select.build(MysqlQueryBuilder);
         debug_print!("{}, {:?}", _sql, _values);
 
         panic!("This is a mock Executor");
     }
+
+    async fn query_all_raw(&self, _sql: String) -> Result<Vec<SqlxRow>, SqlxError> {
+        debug_print!("{}", _sql);
+
+        panic!("This is a mock Executor");
+    }
 }

src/mysql/discovery/executor/real.rs

@@ -1,8 +1,11 @@
 use sea_query::{MysqlQueryBuilder, SelectStatement};
 use sea_query_sqlx::SqlxBinder;
-use sqlx::{MySqlPool, Row, mysql::MySqlRow};
+use sqlx::MySqlPool;
 
-use crate::{debug_print, sqlx_types::SqlxError};
+use crate::{
+    Connection, debug_print,
+    sqlx_types::{Row, SqlxError, SqlxRow, mysql::MySqlRow},
+};
 
 pub struct Executor {
     pool: MySqlPool,
@@ -18,17 +21,6 @@ impl IntoExecutor for MySqlPool {
     }
 }
 
-impl Executor {
-    pub async fn fetch_all(&self, select: SelectStatement) -> Result<Vec<MySqlRow>, SqlxError> {
-        let (sql, values) = select.build_sqlx(MysqlQueryBuilder);
-        debug_print!("{}, {:?}", sql, values);
-
-        sqlx::query_with(&sql, values)
-            .fetch_all(&mut *self.pool.acquire().await?)
-            .await
-    }
-}
-
 pub trait GetMySqlValue {
     fn get_string(&self, idx: usize) -> String;
 
@@ -45,3 +37,29 @@ impl GetMySqlValue for MySqlRow {
             .map(|v| String::from_utf8(v).unwrap())
     }
 }
+
+#[async_trait::async_trait]
+impl Connection for Executor {
+    async fn query_all(&self, select: SelectStatement) -> Result<Vec<SqlxRow>, SqlxError> {
+        let (sql, values) = select.build_sqlx(MysqlQueryBuilder);
+        debug_print!("{}, {:?}", sql, values);
+
+        Ok(sqlx::query_with(&sql, values)
+            .fetch_all(&mut *self.pool.acquire().await?)
+            .await?
+            .into_iter()
+            .map(SqlxRow::MySql)
+            .collect())
+    }
+
+    async fn query_all_raw(&self, sql: String) -> Result<Vec<SqlxRow>, SqlxError> {
+        debug_print!("{}", sql);
+
+        Ok(sqlx::query(&sql)
+            .fetch_all(&mut *self.pool.acquire().await?)
+            .await?
+            .into_iter()
+            .map(SqlxRow::MySql)
+            .collect())
+    }
+}

src/mysql/discovery/mod.rs

@@ -7,41 +7,45 @@ use crate::mysql::query::{
     ColumnQueryResult, ForeignKeyQueryResult, IndexQueryResult, SchemaQueryBuilder,
     TableQueryResult, VersionQueryResult,
 };
-use crate::sqlx_types::SqlxError;
-use futures::future;
+use crate::{
+    Connection,
+    sqlx_types::{MySqlPool, SqlxError},
+};
 use sea_query::{Alias, DynIden, IntoIden, SeaRc};
 
 mod executor;
 pub use executor::*;
 
-pub struct SchemaDiscovery {
+pub struct SchemaDiscovery<C: Connection> {
     pub query: SchemaQueryBuilder,
-    pub executor: Executor,
     pub schema: DynIden,
+    conn: C,
+}
+
+impl SchemaDiscovery<Executor> {
+    /// Discover schema from a SQLx pool
+    pub fn new(pool: MySqlPool, schema: &str) -> Self {
+        Self::conn(pool.into_executor(), schema)
+    }
 }
 
-impl SchemaDiscovery {
-    pub fn new<E>(executor: E, schema: &str) -> Self
-    where
-        E: IntoExecutor,
-    {
-        Self {
+impl<C: Connection> SchemaDiscovery<C> {
+    /// Discover schema from a generic SQLx connection
+    pub fn conn(conn: C, schema: &str) -> Self {
+        SchemaDiscovery {
             query: SchemaQueryBuilder::default(),
-            executor: executor.into_executor(),
             schema: Alias::new(schema).into_iden(),
+            conn,
         }
     }
 
     pub async fn discover(mut self) -> Result<Schema, SqlxError> {
         self.query = SchemaQueryBuilder::new(self.discover_system().await?);
-        let tables = self.discover_tables().await?;
-        let tables = future::try_join_all(
-            tables
-                .into_iter()
-                .map(|t| (&self, t))
-                .map(Self::discover_table_static),
-        )
-        .await?;
+        let mut tables = Vec::new();
+
+        for table in self.discover_tables().await? {
+            tables.push(self.discover_table(table).await?);
+        }
 
         Ok(Schema {
             schema: self.schema.to_string(),
@@ -51,10 +55,10 @@ impl SchemaDiscovery {
     }
 
     pub async fn discover_system(&mut self) -> Result<SystemInfo, SqlxError> {
-        let rows = self.executor.fetch_all(self.query.query_version()).await?;
+        let rows = self.conn.query_all(self.query.query_version()).await?;
 
         #[allow(clippy::never_loop)]
-        for row in rows.iter() {
+        for row in rows {
             let result: VersionQueryResult = row.into();
             debug_print!("{:?}", result);
             let version = result.parse();
@@ -66,12 +70,12 @@ impl SchemaDiscovery {
 
     pub async fn discover_tables(&mut self) -> Result<Vec<TableInfo>, SqlxError> {
         let rows = self
-            .executor
-            .fetch_all(self.query.query_tables(self.schema.clone()))
+            .conn
+            .query_all(self.query.query_tables(self.schema.clone()))
             .await?;
 
         let tables: Vec<TableInfo> = rows
-            .iter()
+            .into_iter()
             .map(|row| {
                 let result: TableQueryResult = row.into();
                 debug_print!("{:?}", result);
@@ -84,12 +88,6 @@ impl SchemaDiscovery {
         Ok(tables)
     }
 
-    async fn discover_table_static(params: (&Self, TableInfo)) -> Result<TableDef, SqlxError> {
-        let this = params.0;
-        let info = params.1;
-        Self::discover_table(this, info).await
-    }
-
     pub async fn discover_table(&self, info: TableInfo) -> Result<TableDef, SqlxError> {
         let table = SeaRc::new(Alias::new(info.name.as_str()));
         let columns = self
@@ -117,12 +115,12 @@ impl SchemaDiscovery {
         system: &SystemInfo,
     ) -> Result<Vec<ColumnInfo>, SqlxError> {
         let rows = self
-            .executor
-            .fetch_all(self.query.query_columns(schema.clone(), table.clone()))
+            .conn
+            .query_all(self.query.query_columns(schema.clone(), table.clone()))
             .await?;
 
         let columns = rows
-            .iter()
+            .into_iter()
             .map(|row| {
                 let result: ColumnQueryResult = row.into();
                 debug_print!("{:?}", result);
@@ -141,12 +139,12 @@ impl SchemaDiscovery {
         table: DynIden,
     ) -> Result<Vec<IndexInfo>, SqlxError> {
         let rows = self
-            .executor
-            .fetch_all(self.query.query_indexes(schema.clone(), table.clone()))
+            .conn
+            .query_all(self.query.query_indexes(schema.clone(), table.clone()))
             .await?;
 
         let results = rows.into_iter().map(|row| {
-            let result: IndexQueryResult = (&row).into();
+            let result: IndexQueryResult = row.into();
             debug_print!("{:?}", result);
             result
         });
@@ -164,12 +162,12 @@ impl SchemaDiscovery {
         table: DynIden,
     ) -> Result<Vec<ForeignKeyInfo>, SqlxError> {
         let rows = self
-            .executor
-            .fetch_all(self.query.query_foreign_key(schema.clone(), table.clone()))
+            .conn
+            .query_all(self.query.query_foreign_key(schema.clone(), table.clone()))
             .await?;
 
         let results = rows.into_iter().map(|row| {
-            let result: ForeignKeyQueryResult = (&row).into();
+            let result: ForeignKeyQueryResult = row.into();
             debug_print!("{:?}", result);
             result
         });

src/mysql/query/column.rs

@@ -1,5 +1,5 @@
 use super::{InformationSchema, SchemaQueryBuilder};
-use crate::sqlx_types::mysql::MySqlRow;
+use crate::sqlx_types::SqlxRow;
 use sea_query::{DynIden, Expr, ExprTrait, Order, Query, SelectStatement, Value};
 
 #[derive(Debug, sea_query::Iden)]
@@ -71,9 +71,10 @@ impl SchemaQueryBuilder {
 }
 
 #[cfg(feature = "sqlx-mysql")]
-impl From<&MySqlRow> for ColumnQueryResult {
-    fn from(row: &MySqlRow) -> Self {
+impl From<SqlxRow> for ColumnQueryResult {
+    fn from(row: SqlxRow) -> Self {
         use crate::mysql::discovery::GetMySqlValue;
+        let row = row.mysql();
         Self {
             column_name: row.get_string(0),
             column_type: row.get_string(1),
@@ -88,8 +89,8 @@ impl From<&MySqlRow> for ColumnQueryResult {
 }
 
 #[cfg(not(feature = "sqlx-mysql"))]
-impl From<&MySqlRow> for ColumnQueryResult {
-    fn from(_: &MySqlRow) -> Self {
+impl From<SqlxRow> for ColumnQueryResult {
+    fn from(_: SqlxRow) -> Self {
         Self::default()
     }
 }

src/mysql/query/foreign_key.rs

@@ -1,5 +1,5 @@
 use super::{InformationSchema, SchemaQueryBuilder};
-use crate::sqlx_types::mysql::MySqlRow;
+use crate::sqlx_types::SqlxRow;
 use sea_query::{DynIden, Expr, ExprTrait, Order, Query, SelectStatement};
 
 #[derive(Debug, sea_query::Iden)]
@@ -79,8 +79,9 @@ impl SchemaQueryBuilder {
 }
 
 #[cfg(feature = "sqlx-mysql")]
-impl From<&MySqlRow> for ForeignKeyQueryResult {
-    fn from(row: &MySqlRow) -> Self {
+impl From<SqlxRow> for ForeignKeyQueryResult {
+    fn from(row: SqlxRow) -> Self {
+        let row = row.mysql();
         use crate::mysql::discovery::GetMySqlValue;
         Self {
             constraint_name: row.get_string(0),
@@ -94,8 +95,8 @@ impl From<&MySqlRow> for ForeignKeyQueryResult {
 }
 
 #[cfg(not(feature = "sqlx-mysql"))]
-impl From<&MySqlRow> for ForeignKeyQueryResult {
-    fn from(_: &MySqlRow) -> Self {
+impl From<SqlxRow> for ForeignKeyQueryResult {
+    fn from(_: SqlxRow) -> Self {
         Self::default()
     }
 }