Add support for running migrations via diesel_migrations

.github/workflows/ci.yml

@@ -4,6 +4,9 @@ on:
   push:
     branches:
       - main
+      - 0.7.x
+      - 0.6.x
+      - 0.5.x
       - 0.3.x
       - 0.4.x
       - 0.2.x
@@ -25,7 +28,13 @@ jobs:
         rust: ["stable"]
         backend: ["postgres", "mysql", "sqlite"]
         os:
-          [ubuntu-latest, macos-13, macos-15, windows-latest, ubuntu-22.04-arm]
+          [
+            ubuntu-latest,
+            macos-15-intel,
+            macos-15,
+            windows-latest,
+            ubuntu-22.04-arm,
+          ]
         include:
           - rust: "beta"
             backend: "postgres"
@@ -100,22 +109,13 @@ jobs:
         run: |
           sudo apt-get update
           sudo apt-get install libsqlite3-dev
-          echo "DATABASE_URL=/tmp/test.db" >> $GITHUB_ENV
+          echo "DATABASE_URL=:memory:" >> $GITHUB_ENV
 
       - name: Install postgres (MacOS)
-        if: matrix.os == 'macos-13' && matrix.backend == 'postgres'
+        if: runner.os == 'macOS' && matrix.backend == 'postgres'
         run: |
           brew install postgresql@14
-          brew services start postgresql@14
-          sleep 3
-          createuser -s postgres
-          echo "DATABASE_URL=postgres://postgres@localhost/" >> $GITHUB_ENV
-
-      - name: Install postgres (MacOS M1)
-        if: matrix.os == 'macos-15' && matrix.backend == 'postgres'
-        run: |
-          brew install postgresql@14
-          brew services start postgresql@14
+          brew services restart postgresql@14
           sleep 3
           createuser -s postgres
           echo "DATABASE_URL=postgres://postgres@localhost/" >> $GITHUB_ENV
@@ -124,10 +124,10 @@ jobs:
         if: runner.os == 'macOS' && matrix.backend == 'sqlite'
         run: |
           brew install sqlite
-          echo "DATABASE_URL=/tmp/test.db" >> $GITHUB_ENV
+          echo "DATABASE_URL=:memory:" >> $GITHUB_ENV
 
-      - name: Install mysql (MacOS)
-        if: matrix.os == 'macos-13' && matrix.backend == 'mysql'
+      - name: Install mysql (MacOS Intel)
+        if: matrix.os == 'macos-15-intel' && matrix.backend == 'mysql'
         run: |
           brew install [email protected]
           /usr/local/opt/[email protected]/bin/mysql_install_db
@@ -184,7 +184,7 @@ jobs:
         run: |
           echo "C:\ProgramData\chocolatey\lib\SQLite\tools" >> $GITHUB_PATH
           echo "SQLITE3_LIB_DIR=C:\ProgramData\chocolatey\lib\SQLite\tools" >> $GITHUB_ENV
-          echo "DATABASE_URL=C:\test.db" >> $GITHUB_ENV
+          echo "DATABASE_URL=:memory:" >> $GITHUB_ENV
 
       - name: Install rust toolchain
         uses: dtolnay/rust-toolchain@master
@@ -194,7 +194,7 @@ jobs:
         run: cargo +${{ matrix.rust }} version
 
       - name: Test diesel_async
-        run: cargo +${{ matrix.rust }} test --manifest-path Cargo.toml --no-default-features --features "${{ matrix.backend }} deadpool bb8 mobc async-connection-wrapper"
+        run: cargo +${{ matrix.rust }} test --manifest-path Cargo.toml --no-default-features --features "${{ matrix.backend }} deadpool bb8 mobc async-connection-wrapper migrations"
 
       - name: Run examples (Postgres)
         if: matrix.backend == 'postgres'

CHANGELOG.md

@@ -6,6 +6,8 @@ for Rust libraries in [RFC #1105](https://github.com/rust-lang/rfcs/blob/master/
 
 ## [Unreleased]
 
+* Added support for running migrations via `AsyncMigrationHarness`
+
 ## [0.6.1] - 2025-07-03
 
 * Fix features for some dependencies

Cargo.toml

@@ -44,6 +44,10 @@ features = [
   "i-implement-a-third-party-backend-and-opt-into-breaking-changes",
 ]
 
+[dependencies.diesel_migrations]
+version = "~2.3.0"
+optional = true
+
 [dev-dependencies]
 tokio = { version = "1.12.0", features = ["rt", "macros", "rt-multi-thread"] }
 cfg-if = "1"
@@ -73,6 +77,7 @@ postgres = ["diesel/postgres_backend", "tokio-postgres", "tokio", "tokio/rt"]
 sqlite = ["diesel/sqlite", "sync-connection-wrapper"]
 sync-connection-wrapper = ["tokio/rt"]
 async-connection-wrapper = ["tokio/net", "tokio/rt"]
+migrations = ["diesel_migrations", "async-connection-wrapper", "tokio/rt-multi-thread"]
 pool = []
 r2d2 = ["pool", "diesel/r2d2"]
 bb8 = ["pool", "dep:bb8"]
@@ -95,10 +100,12 @@ features = [
   "async-connection-wrapper",
   "sync-connection-wrapper",
   "r2d2",
+  "migrations",
+  "tokio/macros",
 ]
 no-default-features = true
 rustc-args = ["--cfg", "docsrs"]
-rustdoc-args = ["--cfg", "docsrs"]
+rustdoc-args = ["--cfg", "docsrs", "-Z", "unstable-options", "--generate-link-to-definition"]
 
 [workspace]
 members = [

examples/postgres/run-pending-migrations-with-rustls/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-diesel-async = { version = "0.6.0", path = "../../../", features = ["bb8", "postgres", "async-connection-wrapper"] }
+diesel-async = { version = "0.6.0", path = "../../../", features = ["bb8", "postgres", "migrations"] }
 futures-util = "0.3.21"
 rustls = "0.23.8"
 rustls-platform-verifier = "0.5.0"

examples/postgres/run-pending-migrations-with-rustls/src/main.rs

@@ -1,6 +1,5 @@
 use diesel::{ConnectionError, ConnectionResult};
-use diesel_async::async_connection_wrapper::AsyncConnectionWrapper;
-use diesel_async::AsyncPgConnection;
+use diesel_async::{AsyncMigrationHarness, AsyncPgConnection};
 use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
 use futures_util::future::BoxFuture;
 use futures_util::FutureExt;
@@ -10,19 +9,15 @@ use rustls_platform_verifier::ConfigVerifierExt;
 pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!();
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
     // Should be in the form of postgres://user:password@localhost/database?sslmode=require
     let db_url = std::env::var("DATABASE_URL").expect("Env var `DATABASE_URL` not set");
 
     let async_connection = establish_connection(db_url.as_str()).await?;
 
-    let mut async_wrapper: AsyncConnectionWrapper<AsyncPgConnection> =
-        AsyncConnectionWrapper::from(async_connection);
-
-    tokio::task::spawn_blocking(move || {
-        async_wrapper.run_pending_migrations(MIGRATIONS).unwrap();
-    })
-    .await?;
+    let mut harness = AsyncMigrationHarness::new(async_connection);
+    harness.run_pending_migrations(MIGRATIONS)?;
+    let _async_connection = harness.into_inner();
 
     Ok(())
 }

src/async_connection_wrapper.rs

@@ -99,7 +99,7 @@ pub type AsyncConnectionWrapper<C, B = self::implementation::Tokio> =
 #[cfg(not(feature = "tokio"))]
 pub use self::implementation::AsyncConnectionWrapper;
 
-mod implementation {
+pub(crate) mod implementation {
     use diesel::connection::{CacheSize, Instrumentation, SimpleConnection};
     use std::ops::{Deref, DerefMut};
 

src/deref_connection.rs

@@ -0,0 +1,150 @@
+use crate::UpdateAndFetchResults;
+use crate::{AsyncConnection, AsyncConnectionCore, SimpleAsyncConnection, TransactionManager};
+use diesel::associations::HasTable;
+use diesel::connection::CacheSize;
+use diesel::connection::Instrumentation;
+use diesel::QueryResult;
+use futures_util::future::BoxFuture;
+use std::ops::DerefMut;
+
+impl<C> SimpleAsyncConnection for C
+where
+    C: DerefMut + Send,
+    C::Target: SimpleAsyncConnection + Send,
+{
+    async fn batch_execute(&mut self, query: &str) -> diesel::QueryResult<()> {
+        let conn = self.deref_mut();
+        conn.batch_execute(query).await
+    }
+}
+
+impl<C> AsyncConnectionCore for C
+where
+    C: DerefMut + Send,
+    C::Target: AsyncConnectionCore,
+{
+    type ExecuteFuture<'conn, 'query> =
+        <C::Target as AsyncConnectionCore>::ExecuteFuture<'conn, 'query>;
+    type LoadFuture<'conn, 'query> = <C::Target as AsyncConnectionCore>::LoadFuture<'conn, 'query>;
+    type Stream<'conn, 'query> = <C::Target as AsyncConnectionCore>::Stream<'conn, 'query>;
+    type Row<'conn, 'query> = <C::Target as AsyncConnectionCore>::Row<'conn, 'query>;
+
+    type Backend = <C::Target as AsyncConnectionCore>::Backend;
+
+    fn load<'conn, 'query, T>(&'conn mut self, source: T) -> Self::LoadFuture<'conn, 'query>
+    where
+        T: diesel::query_builder::AsQuery + 'query,
+        T::Query: diesel::query_builder::QueryFragment<Self::Backend>
+            + diesel::query_builder::QueryId
+            + 'query,
+    {
+        let conn = self.deref_mut();
+        conn.load(source)
+    }
+
+    fn execute_returning_count<'conn, 'query, T>(
+        &'conn mut self,
+        source: T,
+    ) -> Self::ExecuteFuture<'conn, 'query>
+    where
+        T: diesel::query_builder::QueryFragment<Self::Backend>
+            + diesel::query_builder::QueryId
+            + 'query,
+    {
+        let conn = self.deref_mut();
+        conn.execute_returning_count(source)
+    }
+}
+
+#[diagnostic::do_not_recommend]
+impl<C> AsyncConnection for C
+where
+    C: DerefMut + Send,
+    C::Target: AsyncConnection,
+{
+    type TransactionManager =
+        PoolTransactionManager<<C::Target as AsyncConnection>::TransactionManager>;
+
+    async fn establish(_database_url: &str) -> diesel::ConnectionResult<Self> {
+        Err(diesel::result::ConnectionError::BadConnection(
+            String::from("Cannot directly establish a pooled connection"),
+        ))
+    }
+
+    fn transaction_state(
+        &mut self,
+    ) -> &mut <Self::TransactionManager as crate::transaction_manager::TransactionManager<Self>>::TransactionStateData{
+        let conn = self.deref_mut();
+        conn.transaction_state()
+    }
+
+    async fn begin_test_transaction(&mut self) -> diesel::QueryResult<()> {
+        self.deref_mut().begin_test_transaction().await
+    }
+
+    fn instrumentation(&mut self) -> &mut dyn Instrumentation {
+        self.deref_mut().instrumentation()
+    }
+
+    fn set_instrumentation(&mut self, instrumentation: impl Instrumentation) {
+        self.deref_mut().set_instrumentation(instrumentation);
+    }
+
+    fn set_prepared_statement_cache_size(&mut self, size: CacheSize) {
+        self.deref_mut().set_prepared_statement_cache_size(size);
+    }
+}
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct PoolTransactionManager<TM>(std::marker::PhantomData<TM>);
+
+impl<C, TM> TransactionManager<C> for PoolTransactionManager<TM>
+where
+    C: DerefMut + Send,
+    C::Target: AsyncConnection<TransactionManager = TM>,
+    TM: TransactionManager<C::Target>,
+{
+    type TransactionStateData = TM::TransactionStateData;
+
+    async fn begin_transaction(conn: &mut C) -> diesel::QueryResult<()> {
+        TM::begin_transaction(&mut **conn).await
+    }
+
+    async fn rollback_transaction(conn: &mut C) -> diesel::QueryResult<()> {
+        TM::rollback_transaction(&mut **conn).await
+    }
+
+    async fn commit_transaction(conn: &mut C) -> diesel::QueryResult<()> {
+        TM::commit_transaction(&mut **conn).await
+    }
+
+    fn transaction_manager_status_mut(
+        conn: &mut C,
+    ) -> &mut diesel::connection::TransactionManagerStatus {
+        TM::transaction_manager_status_mut(&mut **conn)
+    }
+
+    fn is_broken_transaction_manager(conn: &mut C) -> bool {
+        TM::is_broken_transaction_manager(&mut **conn)
+    }
+}
+
+impl<Changes, Output, Conn> UpdateAndFetchResults<Changes, Output> for Conn
+where
+    Conn: DerefMut + Send,
+    Changes: diesel::prelude::Identifiable + HasTable + Send,
+    Conn::Target: UpdateAndFetchResults<Changes, Output>,
+{
+    fn update_and_fetch<'conn, 'changes>(
+        &'conn mut self,
+        changeset: Changes,
+    ) -> BoxFuture<'changes, QueryResult<Output>>
+    where
+        Changes: 'changes,
+        'conn: 'changes,
+        Self: 'changes,
+    {
+        self.deref_mut().update_and_fetch(changeset)
+    }
+}

src/lib.rs

@@ -5,12 +5,12 @@
 //! with the main diesel crate. It only provides async variants of core diesel traits,
 //! that perform actual io-work.
 //! This includes async counterparts the following traits:
-//! * [`diesel::prelude::RunQueryDsl`](https://docs.diesel.rs/2.0.x/diesel/prelude/trait.RunQueryDsl.html)
-//!    -> [`diesel_async::RunQueryDsl`](crate::RunQueryDsl)
-//! * [`diesel::connection::Connection`](https://docs.diesel.rs/2.0.x/diesel/connection/trait.Connection.html)
-//!    -> [`diesel_async::AsyncConnection`](crate::AsyncConnection)
-//! * [`diesel::query_dsl::UpdateAndFetchResults`](https://docs.diesel.rs/2.0.x/diesel/query_dsl/trait.UpdateAndFetchResults.html)
-//!    -> [`diesel_async::UpdateAndFetchResults`](crate::UpdateAndFetchResults)
+//! * [`diesel::prelude::RunQueryDsl`](https://docs.diesel.rs/2.3.x/diesel/prelude/trait.RunQueryDsl.html)
+//!   -> [`diesel_async::RunQueryDsl`](crate::RunQueryDsl)
+//! * [`diesel::connection::Connection`](https://docs.diesel.rs/2.3.x/diesel/connection/trait.Connection.html)
+//!   -> [`diesel_async::AsyncConnection`](crate::AsyncConnection)
+//! * [`diesel::query_dsl::UpdateAndFetchResults`](https://docs.diesel.rs/2.3.x/diesel/query_dsl/trait.UpdateAndFetchResults.html)
+//!   -> [`diesel_async::UpdateAndFetchResults`](crate::UpdateAndFetchResults)
 //!
 //! These traits closely mirror their diesel counter parts while providing async functionality.
 //!
@@ -65,6 +65,26 @@
 //! #     Ok(())
 //! # }
 //! ```
+//!
+//! ## Crate features:
+//!
+//! * `postgres`: Enables the [`AsyncPgConnection`] implementation
+//! * `mysql`: Enables the [`AsyncMysqlConnection`] implementation
+//! * `sqlite`: Enables the [`SyncConnectionWrapper`](crate::sync_connection_wrapper::SyncConnectionWrapper)
+//!   and everything required to work with SQLite
+//! * `sync-connection-wrapper`: Enables the
+//!   [`SyncConnectionWrapper`](crate::sync_connection_wrapper::SyncConnectionWrapper) which allows to
+//!   wrap sync connections from [`diesel`] into async connection wrapper
+//! * `async-connection-wrapper`: Enables the [`AsyncConnectionWrapper`](crate::async_connection_wrapper::AsyncConnectionWrapper)
+//!   which allows
+//!   to use connection implementations from this crate as sync [`diesel::Connection`]
+//! * `migrations`: Enables the [`AsyncMigrationHarness`] to execute migrations via
+//!   [`diesel_migrations`]
+//! * `pool`: Enables general support for connection pools
+//! * `r2d2`: Enables support for pooling via the [`r2d2`] crate
+//! * `bb8`: Enables support for pooling via the [`bb8`] crate
+//! * `mobc`: Enables support for pooling via the [`mobc`] crate
+//! * `deadpool`: Enables support for pooling via the [`deadpool`] crate
 
 #![warn(
     missing_docs,
@@ -89,6 +109,9 @@ use scoped_futures::{ScopedBoxFuture, ScopedFutureExt};
 
 #[cfg(feature = "async-connection-wrapper")]
 pub mod async_connection_wrapper;
+mod deref_connection;
+#[cfg(feature = "migrations")]
+mod migrations;
 #[cfg(feature = "mysql")]
 mod mysql;
 #[cfg(feature = "postgres")]
@@ -111,6 +134,9 @@ pub use self::pg::AsyncPgConnection;
 #[doc(inline)]
 pub use self::run_query_dsl::*;
 
+#[doc(inline)]
+#[cfg(feature = "migrations")]
+pub use self::migrations::AsyncMigrationHarness;
 #[doc(inline)]
 pub use self::transaction_manager::{AnsiTransactionManager, TransactionManager};