refinery_core/drivers/
mysql_async.rs

1use crate::traits::r#async::{AsyncMigrate, AsyncQuery, AsyncTransaction};
2use crate::util::SchemaVersion;
3use crate::Migration;
4use mysql_async::{
5    prelude::Queryable, Error as MError, IsolationLevel, Pool, Transaction as MTransaction, TxOpts,
6};
7use time::format_description::well_known::Rfc3339;
8use time::OffsetDateTime;
9
10async fn query_applied_migrations<'a>(
11    mut transaction: MTransaction<'a>,
12    query: &str,
13) -> Result<(MTransaction<'a>, Vec<Migration>), MError> {
14    let result = transaction.query(query).await?;
15
16    let applied = result
17        .into_iter()
18        .map(|row| {
19            let (version, name, applied_on, checksum): (SchemaVersion, String, String, String) =
20                mysql_async::from_row(row);
21
22            // Safe to call unwrap, as we stored it in RFC3339 format on the database
23            let applied_on = OffsetDateTime::parse(&applied_on, &Rfc3339).unwrap();
24            Migration::applied(
25                version,
26                name,
27                applied_on,
28                checksum
29                    .parse::<u64>()
30                    .expect("checksum must be a valid u64"),
31            )
32        })
33        .collect();
34
35    Ok((transaction, applied))
36}
37
38impl AsyncTransaction for Pool {
39    type Error = MError;
40
41    async fn execute<'a, T: Iterator<Item = &'a str> + Send + 'a>(
42        &mut self,
43        queries: T,
44    ) -> Result<usize, Self::Error> {
45        let mut conn = self.get_conn().await?;
46        let mut options = TxOpts::new();
47        options.with_isolation_level(Some(IsolationLevel::ReadCommitted));
48
49        let mut transaction = conn.start_transaction(options).await?;
50        let mut count = 0;
51        for query in queries {
52            transaction.query_drop(query).await?;
53            count += 1;
54        }
55        transaction.commit().await?;
56        Ok(count as usize)
57    }
58}
59
60impl AsyncQuery<Vec<Migration>> for Pool {
61    async fn query(
62        &mut self,
63        query: &str,
64    ) -> Result<Vec<Migration>, <Self as AsyncTransaction>::Error> {
65        let mut conn = self.get_conn().await?;
66        let mut options = TxOpts::new();
67        options.with_isolation_level(Some(IsolationLevel::ReadCommitted));
68        let transaction = conn.start_transaction(options).await?;
69
70        let (transaction, applied) = query_applied_migrations(transaction, query).await?;
71        transaction.commit().await?;
72        Ok(applied)
73    }
74}
75
76impl AsyncMigrate for Pool {}
refinery_core/drivers/mysql_async.rs

refinery_core/drivers/
mysql_async.rs
