migra/src/errors.rs

@@ -12,7 +12,6 @@ pub type MigraResult<T> = Result<T, Error>;
 
 /// Migra error
 #[derive(Debug)]
-#[non_exhaustive]
 pub enum Error {
     /// Represents database errors.
     Db(DbError),
@@ -55,7 +54,6 @@ impl Error {
 
 /// All kinds of errors with witch this crate works.
 #[derive(Debug)]
-#[non_exhaustive]
 pub enum DbKind {
     /// Failed to database connection.
     DatabaseConnection,

migra/src/fs.rs

@@ -21,7 +21,7 @@ pub fn get_all_migrations(dir_path: &Path) -> MigraResult<migration::List> {
         Err(e) if e.kind() == io::ErrorKind::NotFound => vec![],
         entries => entries?
             .filter_map(|res| res.ok().map(|e| e.path()))
-            .filter(|path| is_migration_dir(path))
+            .filter(|path| is_migration_dir(&path))
             .collect::<Vec<_>>(),
     };
 

migra/src/migration.rs

@@ -29,6 +29,9 @@ impl Migration {
 ///
 /// Can be presented as a list of all migrations, a list of pending migrations
 /// or a list of applied migrations, depending on the implementation.
+///
+///
+///
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct List {
     inner: Vec<Migration>,
@@ -98,7 +101,7 @@ impl List {
 
     /// Push migration to list.
     pub fn push(&mut self, migration: Migration) {
-        self.inner.push(migration);
+        self.inner.push(migration)
     }
 
     /// Push migration name to list.
@@ -120,7 +123,7 @@ impl List {
     /// # assert_eq!(list, List::from(vec!["name"]));
     /// ```
     pub fn push_name(&mut self, name: &str) {
-        self.inner.push(Migration::new(name));
+        self.inner.push(Migration::new(name))
     }
 
     /// Check if list contains specific migration.
@@ -198,7 +201,7 @@ mod tests {
         assert_eq!(list, List::from(vec![FIRST_MIGRATION]));
 
         list.push(Migration::new(SECOND_MIGRATION));
-        assert_eq!(list, List::from(vec![FIRST_MIGRATION, SECOND_MIGRATION]));
+        assert_eq!(list, List::from(vec![FIRST_MIGRATION, SECOND_MIGRATION]))
     }
 
     #[test]
@@ -209,23 +212,23 @@ mod tests {
         assert_eq!(list, List::from(vec![FIRST_MIGRATION]));
 
         list.push_name(&String::from(SECOND_MIGRATION));
-        assert_eq!(list, List::from(vec![FIRST_MIGRATION, SECOND_MIGRATION]));
+        assert_eq!(list, List::from(vec![FIRST_MIGRATION, SECOND_MIGRATION]))
     }
 
     #[test]
     fn contains_migration() {
         let list = List::from(vec![FIRST_MIGRATION]);
 
-        assert!(list.contains(&Migration::new(FIRST_MIGRATION)));
-        assert!(!list.contains(&Migration::new(SECOND_MIGRATION)));
+        assert_eq!(list.contains(&Migration::new(FIRST_MIGRATION)), true);
+        assert_eq!(list.contains(&Migration::new(SECOND_MIGRATION)), false);
     }
 
     #[test]
     fn contains_migration_name() {
         let list = List::from(vec![FIRST_MIGRATION]);
 
-        assert!(list.contains_name(FIRST_MIGRATION));
-        assert!(!list.contains_name(SECOND_MIGRATION));
+        assert_eq!(list.contains_name(FIRST_MIGRATION), true);
+        assert_eq!(list.contains_name(SECOND_MIGRATION), false);
     }
 
     #[test]
@@ -234,6 +237,6 @@ mod tests {
         let applied_migrations = List::from(vec![FIRST_MIGRATION]);
         let excluded = all_migrations.exclude(&applied_migrations);
 
-        assert_eq!(excluded, List::from(vec![SECOND_MIGRATION]));
+        assert_eq!(excluded, List::from(vec![SECOND_MIGRATION]))
     }
 }

migra_cli/src/commands/apply.rs

@@ -20,12 +20,22 @@ pub(crate) fn apply_sql(app: &App, cmd_opts: &ApplyCommandOpt) -> migra::StdResu
         .map(std::fs::read_to_string)
         .collect::<Result<Vec<_>, _>>()?;
 
-    database::run_in_transaction(&mut client, |client| {
-        file_contents
-            .iter()
-            .try_for_each(|content| client.apply_sql(content))
-            .map_err(From::from)
-    })?;
+    database::should_run_in_transaction(
+        &mut client,
+        cmd_opts.transaction_opts.single_transaction,
+        |client| {
+            file_contents
+                .iter()
+                .try_for_each(|content| {
+                    database::should_run_in_transaction(
+                        client,
+                        !cmd_opts.transaction_opts.single_transaction,
+                        |client| client.apply_sql(content),
+                    )
+                })
+                .map_err(From::from)
+        },
+    )?;
 
     Ok(())
 }

migra_cli/src/commands/downgrade.rs

@@ -32,19 +32,27 @@ pub(crate) fn rollback_applied_migrations(
         })
         .collect::<Result<Vec<_>, _>>()?;
 
-    database::run_in_transaction(&mut client, |client| {
-        migrations_with_content
-            .iter()
-            .try_for_each(|(migration_name, content)| {
-                if all_migrations.contains_name(migration_name) {
-                    println!("downgrade {}...", migration_name);
-                    client.run_downgrade_migration(migration_name, content)
-                } else {
-                    Ok(())
-                }
-            })
-            .map_err(From::from)
-    })?;
+    database::should_run_in_transaction(
+        &mut client,
+        opts.transaction_opts.single_transaction,
+        |client| {
+            migrations_with_content
+                .iter()
+                .try_for_each(|(migration_name, content)| {
+                    if all_migrations.contains_name(migration_name) {
+                        println!("downgrade {}...", migration_name);
+                        database::should_run_in_transaction(
+                            client,
+                            !opts.transaction_opts.single_transaction,
+                            |client| client.run_downgrade_migration(migration_name, &content),
+                        )
+                    } else {
+                        Ok(())
+                    }
+                })
+                .map_err(From::from)
+        },
+    )?;
 
     Ok(())
 }

migra_cli/src/commands/upgrade.rs

@@ -52,15 +52,23 @@ pub(crate) fn upgrade_pending_migrations(
         })
         .collect::<Result<Vec<_>, _>>()?;
 
-    database::run_in_transaction(&mut client, |client| {
-        migrations_with_content
-            .iter()
-            .try_for_each(|(migration_name, content)| {
-                println!("upgrade {}...", migration_name);
-                client.run_upgrade_migration(migration_name, content)
-            })
-            .map_err(From::from)
-    })?;
+    database::should_run_in_transaction(
+        &mut client,
+        opts.transaction_opts.single_transaction,
+        |client| {
+            migrations_with_content
+                .iter()
+                .try_for_each(|(migration_name, content)| {
+                    println!("upgrade {}...", migration_name);
+                    database::should_run_in_transaction(
+                        client,
+                        !opts.transaction_opts.single_transaction,
+                        |client| client.run_upgrade_migration(migration_name, &content),
+                    )
+                })
+                .map_err(From::from)
+        },
+    )?;
 
     Ok(())
 }

migra_cli/src/config.rs

@@ -50,7 +50,7 @@ fn is_sqlite_database_file(filename: &str) -> bool {
         .rsplit('.')
         .next()
         .map(|ext| ext.eq_ignore_ascii_case("db"))
-        .unwrap_or_default()
+        == Some(true)
 }
 
 fn default_database_connection_env() -> String {
@@ -138,7 +138,7 @@ impl DatabaseConfig {
     }
 
     pub fn connection_string(&self) -> MigraResult<String> {
-        self.connection.strip_prefix('$').map_or_else(
+        self.connection.strip_prefix("$").map_or_else(
             || Ok(self.connection.clone()),
             |connection_env| {
                 env::var(connection_env)
@@ -183,7 +183,7 @@ impl Default for MigrationsConfig {
 
 impl MigrationsConfig {
     pub fn directory(&self) -> String {
-        self.directory.strip_prefix('$').map_or_else(
+        self.directory.strip_prefix("$").map_or_else(
             || self.directory.clone(),
             |directory_env| {
                 env::var(directory_env).unwrap_or_else(|_| {
@@ -199,7 +199,7 @@ impl MigrationsConfig {
     }
 
     pub fn table_name(&self) -> String {
-        self.table_name.strip_prefix('$').map_or_else(
+        self.table_name.strip_prefix("$").map_or_else(
             || self.table_name.clone(),
             |table_name_env| {
                 env::var(table_name_env).unwrap_or_else(|_| {

migra_cli/src/database.rs

@@ -53,3 +53,18 @@ where
         .and_then(|res| client.commit_transaction().and(Ok(res)))
         .or_else(|err| client.rollback_transaction().and(Err(err)))
 }
+
+pub fn should_run_in_transaction<TrxFnMut>(
+    client: &mut AnyClient,
+    is_needed: bool,
+    trx_fn: TrxFnMut,
+) -> migra::Result<()>
+where
+    TrxFnMut: FnOnce(&mut AnyClient) -> migra::Result<()>,
+{
+    if is_needed {
+        run_in_transaction(client, trx_fn)
+    } else {
+        trx_fn(client)
+    }
+}

migra_cli/src/opts.rs

@@ -32,10 +32,19 @@ pub(crate) enum Command {
     Completions(CompletionsShell),
 }
 
+#[derive(Debug, StructOpt, Clone)]
+pub(crate) struct TransactionOpts {
+    #[structopt(long = "single-transaction")]
+    pub single_transaction: bool,
+}
+
 #[derive(Debug, StructOpt, Clone)]
 pub(crate) struct ApplyCommandOpt {
     #[structopt(parse(from_os_str), required = true)]
     pub file_paths: Vec<PathBuf>,
+
+    #[structopt(flatten)]
+    pub transaction_opts: TransactionOpts,
 }
 
 #[derive(Debug, StructOpt, Clone)]
@@ -55,6 +64,9 @@ pub(crate) struct UpgradeCommandOpt {
     /// How many existing migrations do we have to update.
     #[structopt(long = "number", short = "n")]
     pub migrations_number: Option<usize>,
+
+    #[structopt(flatten)]
+    pub transaction_opts: TransactionOpts,
 }
 
 #[derive(Debug, StructOpt, Clone)]
@@ -66,6 +78,9 @@ pub(crate) struct DowngradeCommandOpt {
     /// Rolls back all applied migrations. Ignores --number option.
     #[structopt(long = "all")]
     pub all_migrations: bool,
+
+    #[structopt(flatten)]
+    pub transaction_opts: TransactionOpts,
 }
 
 #[derive(Debug, StructOpt, Clone)]

migra_cli/tests/commands.rs

@@ -485,6 +485,62 @@ mod upgrade {
         Ok(())
     }
 
+    #[test]
+    fn partial_applied_invalid_migrations() -> TestResult {
+        fn inner<ValidateFn>(database_name: &'static str, validate: ValidateFn) -> TestResult
+        where
+            ValidateFn: Fn() -> TestResult,
+        {
+            let manifest_path = database_manifest_path(database_name);
+
+            Command::cargo_bin("migra")?
+                .arg("-c")
+                .arg(&manifest_path)
+                .arg("up")
+                .assert()
+                .failure();
+
+            validate()?;
+
+            Command::cargo_bin("migra")?
+                .arg("-c")
+                .arg(&manifest_path)
+                .arg("down")
+                .assert()
+                .success();
+
+            Ok(())
+        }
+
+        #[cfg(feature = "postgres")]
+        inner("postgres_invalid", || {
+            let mut conn = client_postgres::Client::connect(POSTGRES_URL, client_postgres::NoTls)?;
+            let articles_res = conn.query("SELECT a.id FROM articles AS a", &[]);
+            let persons_res = conn.query("SELECT p.id FROM persons AS p", &[]);
+
+            assert!(articles_res.is_ok());
+            assert!(persons_res.is_err());
+
+            Ok(())
+        })?;
+
+        #[cfg(feature = "sqlite")]
+        remove_sqlite_db().and_then(|_| {
+            inner("sqlite_invalid", || {
+                let conn = client_rusqlite::Connection::open(SQLITE_URL)?;
+                let articles_res = conn.execute_batch("SELECT a.id FROM articles AS a");
+                let persons_res = conn.execute_batch("SELECT p.id FROM persons AS p");
+
+                assert!(articles_res.is_ok());
+                assert!(persons_res.is_err());
+
+                Ok(())
+            })
+        })?;
+
+        Ok(())
+    }
+
     #[test]
     fn cannot_applied_invalid_migrations_in_single_transaction() -> TestResult {
         fn inner<ValidateFn>(database_name: &'static str, validate: ValidateFn) -> TestResult
@@ -532,8 +588,6 @@ mod upgrade {
             })
         })?;
 
-        // mysql doesn't support DDL in transaction 🤷
-
         Ok(())
     }
 }