tern_core/executor/sqlx_backend/
pool.rs

1//! [`Executor`] for the generic [`sqlx::Pool`][sqlx-pool], a pool of `sqlx`
2//! database connections.
3//!
4//! [`Executor`]: crate::migration::Executor
5//! [sqlx-pool]: https://docs.rs/sqlx/0.8.3/sqlx/struct.Pool.html
6use crate::error::{DatabaseError as _, TernResult};
7use crate::migration::{
8    AppliedMigration, Executor as MigrationExecutor, Query, QueryRepository,
9};
10
11use chrono::{DateTime, Utc};
12use sqlx::pool::PoolOptions;
13use sqlx::{
14    Acquire, Connection, Database, Encode, Executor, FromRow, IntoArguments,
15    Pool, Type,
16};
17use std::marker::PhantomData;
18
19/// The generic `sqlx::Pool` as a migration executor backend.
20pub struct SqlxExecutor<Db, Q>
21where
22    Db: Database,
23    Q: QueryRepository,
24{
25    pool: Pool<Db>,
26    _q: PhantomData<Q>,
27}
28
29impl<Db, Q> SqlxExecutor<Db, Q>
30where
31    Db: Database,
32    Q: QueryRepository,
33{
34    /// Create a pool with default options from a connection string.
35    pub async fn new(db_url: &str) -> TernResult<Self> {
36        let pool = Pool::connect(db_url).await.tern_result()?;
37
38        Ok(Self { pool, _q: PhantomData })
39    }
40
41    /// Create the pool from the given options.
42    pub async fn new_with(
43        pool_opts: PoolOptions<Db>,
44        conn_opts: <Db::Connection as Connection>::Options,
45    ) -> TernResult<Self> {
46        let pool = pool_opts.connect_with(conn_opts).await.tern_result()?;
47
48        Ok(Self { pool, _q: PhantomData })
49    }
50
51    /// Exposing the underlying connection object for usage involving queries
52    /// beyond what `Executor` details.
53    pub fn pool(&self) -> Pool<Db> {
54        self.pool.clone()
55    }
56}
57
58/// `SqlxExecutor` can be an [`Executor`] fairly straightforwardly when enough
59/// bounds involving `Db: sqlx::Database` are added to make it compile.
60///
61/// [`Executor`]: crate::migration::Executor
62impl<Db, Q> MigrationExecutor for SqlxExecutor<Db, Q>
63where
64    Self: Send + Sync + 'static,
65    Q: QueryRepository,
66    Db: Database,
67    for<'c> &'c mut <Db as Database>::Connection: Executor<'c, Database = Db>,
68    for<'q> <Db as Database>::Arguments<'q>: IntoArguments<'q, Db>,
69    for<'r> AppliedMigration: FromRow<'r, <Db as Database>::Row>,
70    String: Type<Db> + for<'a> Encode<'a, Db>,
71    i64: Type<Db> + for<'a> Encode<'a, Db>,
72    DateTime<Utc>: Type<Db> + for<'a> Encode<'a, Db>,
73{
74    type Queries = Q;
75
76    async fn apply_tx(&mut self, query: &Query) -> TernResult<()> {
77        let mut tx = self.pool.begin().await.tern_result()?;
78        let conn = tx.acquire().await.tern_result()?;
79        conn.execute(sqlx::raw_sql(query.sql())).await.void_tern_result()?;
80        tx.commit().await.void_tern_result()?;
81
82        Ok(())
83    }
84
85    async fn apply_no_tx(&mut self, query: &Query) -> TernResult<()> {
86        let statements = query.split_statements()?;
87        for statement in statements.iter() {
88            self.pool
89                .execute(sqlx::raw_sql(statement.as_ref()))
90                .await
91                .void_tern_result()?;
92        }
93
94        Ok(())
95    }
96
97    async fn create_history_if_not_exists(
98        &mut self,
99        history_table: &str,
100    ) -> TernResult<()> {
101        let query = Q::create_history_if_not_exists_query(history_table);
102        self.pool.execute(sqlx::raw_sql(query.sql())).await.void_tern_result()
103    }
104
105    async fn drop_history(&mut self, history_table: &str) -> TernResult<()> {
106        let query = Q::drop_history_query(history_table);
107        self.pool.execute(sqlx::raw_sql(query.sql())).await.void_tern_result()
108    }
109
110    async fn get_all_applied(
111        &mut self,
112        history_table: &str,
113    ) -> TernResult<Vec<AppliedMigration>> {
114        let query = Q::select_star_from_history_query(history_table);
115        let applied = sqlx::query_as::<Db, AppliedMigration>(query.sql())
116            .fetch_all(&self.pool)
117            .await
118            .tern_result()?;
119
120        Ok(applied)
121    }
122
123    /// This expects [`insert_into_history_query`] to have placeholders for
124    /// `bind`ing the fields of the `AppliedMigration`, and that they appear in
125    /// the same order as they do in the [`AppliedMigration`] struct.
126    ///
127    /// [`insert_into_history_query`]: crate::migration::QueryRepository::insert_into_history_query
128    /// [`AppliedMigration`]: crate::migration::AppliedMigration
129    async fn insert_applied_migration(
130        &mut self,
131        history_table: &str,
132        applied: &AppliedMigration,
133    ) -> TernResult<()> {
134        let query = Q::insert_into_history_query(history_table, applied);
135        sqlx::query::<Db>(query.sql())
136            .bind(applied.version)
137            .bind(applied.description.clone())
138            .bind(applied.content.clone())
139            .bind(applied.duration_ms)
140            .bind(applied.applied_at)
141            .execute(&self.pool)
142            .await
143            .void_tern_result()?;
144
145        Ok(())
146    }
147
148    /// Like [`insert_applied_migration`] this expects a query with placeholders
149    /// lining up with the order of [`AppliedMigration`] fields.
150    ///
151    /// [`insert_applied_migration`]: Self::insert_applied_migration
152    /// [`AppliedMigration`]: crate::migration::AppliedMigration
153    async fn upsert_applied_migration(
154        &mut self,
155        history_table: &str,
156        applied: &AppliedMigration,
157    ) -> TernResult<()> {
158        let query = Q::upsert_history_query(history_table, applied);
159        sqlx::query::<Db>(query.sql())
160            .bind(applied.version)
161            .bind(applied.description.clone())
162            .bind(applied.content.clone())
163            .bind(applied.duration_ms)
164            .bind(applied.applied_at)
165            .execute(&self.pool)
166            .await
167            .void_tern_result()?;
168
169        Ok(())
170    }
171}
tern_core/executor/sqlx_backend/pool.rs

tern_core/executor/sqlx_backend/
pool.rs
