Close connection and transaction

src/database/connection.rs

@@ -31,6 +31,9 @@ pub trait ConnectionTrait: Sync {
     fn is_mock_connection(&self) -> bool {
         false
     }
+
+    /// Explicitly close the database connection
+    async fn close(self) -> Result<(), DbErr>;
 }
 
 /// Stream query results

src/database/db_connection.rs

@@ -162,6 +162,25 @@ impl ConnectionTrait for DatabaseConnection {
     fn is_mock_connection(&self) -> bool {
         matches!(self, DatabaseConnection::MockDatabaseConnection(_))
     }
+
+    async fn close(self) -> Result<(), DbErr> {
+        match self {
+            #[cfg(feature = "sqlx-mysql")]
+            DatabaseConnection::SqlxMySqlPoolConnection(conn) => conn.close().await,
+            #[cfg(feature = "sqlx-postgres")]
+            DatabaseConnection::SqlxPostgresPoolConnection(conn) => conn.close().await,
+            #[cfg(feature = "sqlx-sqlite")]
+            DatabaseConnection::SqlxSqlitePoolConnection(conn) => conn.close().await,
+            #[cfg(feature = "mock")]
+            DatabaseConnection::MockDatabaseConnection(conn) => {
+                // Nothing to cleanup, we just consume the `DatabaseConnection`
+                Ok(())
+            }
+            DatabaseConnection::Disconnected => {
+                Err(DbErr::Conn(RuntimeErr::Internal("Disconnected".to_owned())))
+            }
+        }
+    }
 }
 
 #[async_trait::async_trait]

src/database/transaction.rs

@@ -386,6 +386,10 @@ impl ConnectionTrait for DatabaseTransaction {
             _ => unreachable!(),
         }
     }
+
+    async fn close(self) -> Result<(), DbErr> {
+        self.rollback().await
+    }
 }
 
 impl StreamTrait for DatabaseTransaction {

src/driver/sqlx_mysql.rs

@@ -185,6 +185,11 @@ impl SqlxMySqlPoolConnection {
     {
         self.metric_callback = Some(Arc::new(callback));
     }
+
+    /// Explicitly close the MySQL connection
+    pub async fn close(self) -> Result<(), DbErr> {
+        Ok(self.pool.close().await)
+    }
 }
 
 impl From<MySqlRow> for QueryResult {

src/driver/sqlx_postgres.rs

@@ -200,6 +200,11 @@ impl SqlxPostgresPoolConnection {
     {
         self.metric_callback = Some(Arc::new(callback));
     }
+
+    /// Explicitly close the Postgres connection
+    pub async fn close(self) -> Result<(), DbErr> {
+        Ok(self.pool.close().await)
+    }
 }
 
 impl From<PgRow> for QueryResult {

src/driver/sqlx_sqlite.rs

@@ -192,6 +192,11 @@ impl SqlxSqlitePoolConnection {
     {
         self.metric_callback = Some(Arc::new(callback));
     }
+
+    /// Explicitly close the SQLite connection
+    pub async fn close(self) -> Result<(), DbErr> {
+        Ok(self.pool.close().await)
+    }
 }
 
 impl From<SqliteRow> for QueryResult {