refinery_core/drivers/
tiberius.rs

1use crate::traits::r#async::{AsyncMigrate, AsyncQuery, AsyncTransaction};
2use crate::util::SchemaVersion;
3use crate::Migration;
4
5use futures::{
6    io::{AsyncRead, AsyncWrite},
7    TryStreamExt,
8};
9use tiberius::{error::Error, Client, QueryItem};
10use time::format_description::well_known::Rfc3339;
11use time::OffsetDateTime;
12
13async fn query_applied_migrations<S: AsyncRead + AsyncWrite + Unpin + Send>(
14    client: &mut Client<S>,
15    query: &str,
16) -> Result<Vec<Migration>, Error> {
17    let mut rows = client.simple_query(query).await?;
18    let mut applied = Vec::new();
19    // Unfortunately too many unwraps as `Row::get` maps to Option<T> instead of T
20    while let Some(item) = rows.try_next().await? {
21        if let QueryItem::Row(row) = item {
22            let version = row.get::<SchemaVersion, usize>(0).unwrap();
23            let applied_on: &str = row.get::<&str, usize>(2).unwrap();
24            // Safe to call unwrap, as we stored it in RFC3339 format on the database
25            let applied_on = OffsetDateTime::parse(applied_on, &Rfc3339).unwrap();
26            let checksum: String = row.get::<&str, usize>(3).unwrap().to_string();
27
28            applied.push(Migration::applied(
29                version,
30                row.get::<&str, usize>(1).unwrap().to_string(),
31                applied_on,
32                checksum
33                    .parse::<u64>()
34                    .expect("checksum must be a valid u64"),
35            ));
36        }
37    }
38
39    Ok(applied)
40}
41
42impl<S> AsyncTransaction for Client<S>
43where
44    S: AsyncRead + AsyncWrite + Unpin + Send,
45{
46    type Error = Error;
47
48    async fn execute<'a, T: Iterator<Item = &'a str> + Send + 'a>(
49        &'a mut self,
50        queries: T,
51    ) -> Result<usize, Self::Error> {
52        // Tiberius doesn't support transactions, see https://github.com/prisma/tiberius/issues/28
53        self.simple_query("BEGIN TRAN T1;").await?;
54        let mut count = 0;
55        for query in queries {
56            // Drop the returning `QueryStream<'a>` to avoid compiler complaning regarding lifetimes
57            if let Err(err) = self.simple_query(query).await.map(drop) {
58                if let Err(err) = self.simple_query("ROLLBACK TRAN T1").await {
59                    log::error!("could not ROLLBACK transaction, {}", err);
60                }
61                return Err(err);
62            }
63            count += 1;
64        }
65        self.simple_query("COMMIT TRAN T1").await?;
66        Ok(count as usize)
67    }
68}
69
70impl<S> AsyncQuery<Vec<Migration>> for Client<S>
71where
72    S: AsyncRead + AsyncWrite + Unpin + Send,
73{
74    async fn query<'a>(
75        &'a mut self,
76        query: &'a str,
77    ) -> Result<Vec<Migration>, <Self as AsyncTransaction>::Error> {
78        let applied = query_applied_migrations(self, query).await?;
79        Ok(applied)
80    }
81}
82
83impl<S> AsyncMigrate for Client<S>
84where
85    S: AsyncRead + AsyncWrite + Unpin + Send,
86{
87    fn assert_migrations_table_query(migration_table_name: &str) -> String {
88        format!(
89            "IF NOT EXISTS(SELECT 1 FROM sys.Tables WHERE  Name = N'{migration_table_name}')
90         BEGIN
91           CREATE TABLE {migration_table_name}(
92             version INT PRIMARY KEY,
93             name VARCHAR(255),
94             applied_on VARCHAR(255),
95             checksum VARCHAR(255));
96         END"
97        )
98    }
99}
refinery_core/drivers/tiberius.rs

refinery_core/drivers/
tiberius.rs
