From a2f5e8101aa18569286f6590a0820541f4f0c9dc Mon Sep 17 00:00:00 2001 From: Billy Chan Date: Mon, 9 Jan 2023 23:21:26 +0800 Subject: [PATCH 1/9] Running atomic migration --- sea-orm-migration/Cargo.toml | 1 + sea-orm-migration/src/lib.rs | 2 + sea-orm-migration/src/manager.rs | 20 ++- sea-orm-migration/src/manager_connection.rs | 145 ++++++++++++++++++ sea-orm-migration/src/migrator.rs | 115 ++++++++++---- sea-orm-migration/src/prelude.rs | 19 +-- sea-orm-migration/tests/main.rs | 134 +++++++++------- .../m20230109_000001_seed_cake_table.rs | 57 +++++++ sea-orm-migration/tests/migrator/mod.rs | 2 + 9 files changed, 400 insertions(+), 95 deletions(-) create mode 100644 sea-orm-migration/src/manager_connection.rs create mode 100644 sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs diff --git a/sea-orm-migration/Cargo.toml b/sea-orm-migration/Cargo.toml index bb8e5e889..555849f6b 100644 --- a/sea-orm-migration/Cargo.toml +++ b/sea-orm-migration/Cargo.toml @@ -28,6 +28,7 @@ sea-orm-cli = { version = "0.10.3", path = "../sea-orm-cli", default-features = sea-schema = { version = "0.11" } tracing = { version = "0.1", default-features = false, features = ["log"] } tracing-subscriber = { version = "0.3", default-features = false, features = ["env-filter", "fmt"] } +futures = { version = "0.3", default-features = false, features = ["std"] } [dev-dependencies] async-std = { version = "1", features = ["attributes", "tokio1"] } diff --git a/sea-orm-migration/src/lib.rs b/sea-orm-migration/src/lib.rs index 6a7f8547f..38b0dd1bc 100644 --- a/sea-orm-migration/src/lib.rs +++ b/sea-orm-migration/src/lib.rs @@ -1,12 +1,14 @@ #[cfg(feature = "cli")] pub mod cli; pub mod manager; +pub mod manager_connection; pub mod migrator; pub mod prelude; pub mod seaql_migrations; pub mod util; pub use manager::*; +pub use manager_connection::*; pub use migrator::*; pub use async_trait; diff --git a/sea-orm-migration/src/manager.rs b/sea-orm-migration/src/manager.rs index 71e91b379..88b1235d1 100644 --- a/sea-orm-migration/src/manager.rs +++ b/sea-orm-migration/src/manager.rs @@ -1,20 +1,28 @@ +use super::{IntoSchemaManagerConnection, SchemaManagerConnection}; use sea_orm::sea_query::{ extension::postgres::{TypeAlterStatement, TypeCreateStatement, TypeDropStatement}, ForeignKeyCreateStatement, ForeignKeyDropStatement, IndexCreateStatement, IndexDropStatement, TableAlterStatement, TableCreateStatement, TableDropStatement, TableRenameStatement, TableTruncateStatement, }; -use sea_orm::{ConnectionTrait, DbBackend, DbConn, DbErr, StatementBuilder}; +use sea_orm::{ + ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbBackend, DbErr, StatementBuilder, +}; use sea_schema::{mysql::MySql, postgres::Postgres, probe::SchemaProbe, sqlite::Sqlite}; /// Helper struct for writing migration scripts in migration file pub struct SchemaManager<'c> { - conn: &'c DbConn, + conn: SchemaManagerConnection<'c>, } impl<'c> SchemaManager<'c> { - pub fn new(conn: &'c DbConn) -> Self { - Self { conn } + pub fn new(conn: T) -> Self + where + T: IntoSchemaManagerConnection<'c>, + { + Self { + conn: conn.into_schema_manager_connection(), + } } pub async fn exec_stmt(&self, stmt: S) -> Result<(), DbErr> @@ -29,8 +37,8 @@ impl<'c> SchemaManager<'c> { self.conn.get_database_backend() } - pub fn get_connection(&self) -> &'c DbConn { - self.conn + pub fn get_connection(&self) -> &SchemaManagerConnection<'c> { + &self.conn } } diff --git a/sea-orm-migration/src/manager_connection.rs b/sea-orm-migration/src/manager_connection.rs new file mode 100644 index 000000000..eda6f6284 --- /dev/null +++ b/sea-orm-migration/src/manager_connection.rs @@ -0,0 +1,145 @@ +use futures::Future; +use sea_orm::{ + AccessMode, ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbBackend, DbErr, + ExecResult, IsolationLevel, QueryResult, Statement, TransactionError, TransactionTrait, +}; +use std::pin::Pin; + +pub enum SchemaManagerConnection<'c> { + Connection(&'c DatabaseConnection), + Transaction(&'c DatabaseTransaction), +} + +#[async_trait::async_trait] +impl<'c> ConnectionTrait for SchemaManagerConnection<'c> { + fn get_database_backend(&self) -> DbBackend { + match self { + SchemaManagerConnection::Connection(conn) => conn.get_database_backend(), + SchemaManagerConnection::Transaction(trans) => trans.get_database_backend(), + } + } + + async fn execute(&self, stmt: Statement) -> Result { + match self { + SchemaManagerConnection::Connection(conn) => conn.execute(stmt).await, + SchemaManagerConnection::Transaction(trans) => trans.execute(stmt).await, + } + } + + async fn execute_unprepared(&self, sql: &str) -> Result { + match self { + SchemaManagerConnection::Connection(conn) => conn.execute_unprepared(sql).await, + SchemaManagerConnection::Transaction(trans) => trans.execute_unprepared(sql).await, + } + } + + async fn query_one(&self, stmt: Statement) -> Result, DbErr> { + match self { + SchemaManagerConnection::Connection(conn) => conn.query_one(stmt).await, + SchemaManagerConnection::Transaction(trans) => trans.query_one(stmt).await, + } + } + + async fn query_all(&self, stmt: Statement) -> Result, DbErr> { + match self { + SchemaManagerConnection::Connection(conn) => conn.query_all(stmt).await, + SchemaManagerConnection::Transaction(trans) => trans.query_all(stmt).await, + } + } + + fn is_mock_connection(&self) -> bool { + match self { + SchemaManagerConnection::Connection(conn) => conn.is_mock_connection(), + SchemaManagerConnection::Transaction(trans) => trans.is_mock_connection(), + } + } +} + +#[async_trait::async_trait] +impl<'c> TransactionTrait for SchemaManagerConnection<'c> { + async fn begin(&self) -> Result { + match self { + SchemaManagerConnection::Connection(conn) => conn.begin().await, + SchemaManagerConnection::Transaction(trans) => trans.begin().await, + } + } + + async fn begin_with_config( + &self, + isolation_level: Option, + access_mode: Option, + ) -> Result { + match self { + SchemaManagerConnection::Connection(conn) => { + conn.begin_with_config(isolation_level, access_mode).await + } + SchemaManagerConnection::Transaction(trans) => { + trans.begin_with_config(isolation_level, access_mode).await + } + } + } + + async fn transaction(&self, callback: F) -> Result> + where + F: for<'a> FnOnce( + &'a DatabaseTransaction, + ) -> Pin> + Send + 'a>> + + Send, + T: Send, + E: std::error::Error + Send, + { + match self { + SchemaManagerConnection::Connection(conn) => conn.transaction(callback).await, + SchemaManagerConnection::Transaction(trans) => trans.transaction(callback).await, + } + } + + async fn transaction_with_config( + &self, + callback: F, + isolation_level: Option, + access_mode: Option, + ) -> Result> + where + F: for<'a> FnOnce( + &'a DatabaseTransaction, + ) -> Pin> + Send + 'a>> + + Send, + T: Send, + E: std::error::Error + Send, + { + match self { + SchemaManagerConnection::Connection(conn) => { + conn.transaction_with_config(callback, isolation_level, access_mode) + .await + } + SchemaManagerConnection::Transaction(trans) => { + trans + .transaction_with_config(callback, isolation_level, access_mode) + .await + } + } + } +} + +pub trait IntoSchemaManagerConnection<'c>: Send { + fn into_schema_manager_connection(self) -> SchemaManagerConnection<'c>; +} + +impl<'c> IntoSchemaManagerConnection<'c> for SchemaManagerConnection<'c> { + fn into_schema_manager_connection(self) -> SchemaManagerConnection<'c> { + self + } +} + +impl<'c> IntoSchemaManagerConnection<'c> for &'c DatabaseConnection { + fn into_schema_manager_connection(self) -> SchemaManagerConnection<'c> { + SchemaManagerConnection::Connection(self) + } +} + +impl<'c> IntoSchemaManagerConnection<'c> for &'c DatabaseTransaction { + fn into_schema_manager_connection(self) -> SchemaManagerConnection<'c> { + SchemaManagerConnection::Transaction(self) + } +} diff --git a/sea-orm-migration/src/migrator.rs b/sea-orm-migration/src/migrator.rs index 632ef0c83..cf54aa23f 100644 --- a/sea-orm-migration/src/migrator.rs +++ b/sea-orm-migration/src/migrator.rs @@ -13,7 +13,7 @@ use sea_orm::{ }; use sea_schema::{mysql::MySql, postgres::Postgres, probe::SchemaProbe, sqlite::Sqlite}; -use super::{seaql_migrations, MigrationTrait, SchemaManager}; +use super::{seaql_migrations, IntoSchemaManagerConnection, MigrationTrait, SchemaManager}; #[derive(Debug, PartialEq, Eq)] /// Status of migration @@ -57,7 +57,10 @@ pub trait MigratorTrait: Send { } /// Get list of applied migrations from database - async fn get_migration_models(db: &DbConn) -> Result, DbErr> { + async fn get_migration_models(db: &C) -> Result, DbErr> + where + C: ConnectionTrait, + { Self::install(db).await?; seaql_migrations::Entity::find() .order_by_asc(seaql_migrations::Column::Version) @@ -66,7 +69,10 @@ pub trait MigratorTrait: Send { } /// Get list of migrations with status - async fn get_migration_with_status(db: &DbConn) -> Result, DbErr> { + async fn get_migration_with_status(db: &C) -> Result, DbErr> + where + C: ConnectionTrait, + { Self::install(db).await?; let mut migration_files = Self::get_migration_files(); let migration_models = Self::get_migration_models(db).await?; @@ -102,7 +108,10 @@ pub trait MigratorTrait: Send { } /// Get list of pending migrations - async fn get_pending_migrations(db: &DbConn) -> Result, DbErr> { + async fn get_pending_migrations(db: &C) -> Result, DbErr> + where + C: ConnectionTrait, + { Self::install(db).await?; Ok(Self::get_migration_with_status(db) .await? @@ -112,7 +121,10 @@ pub trait MigratorTrait: Send { } /// Get list of applied migrations - async fn get_applied_migrations(db: &DbConn) -> Result, DbErr> { + async fn get_applied_migrations(db: &C) -> Result, DbErr> + where + C: ConnectionTrait, + { Self::install(db).await?; Ok(Self::get_migration_with_status(db) .await? @@ -122,7 +134,10 @@ pub trait MigratorTrait: Send { } /// Create migration table `seaql_migrations` in the database - async fn install(db: &DbConn) -> Result<(), DbErr> { + async fn install(db: &C) -> Result<(), DbErr> + where + C: ConnectionTrait, + { let builder = db.get_database_backend(); let schema = Schema::new(builder); let mut stmt = schema.create_table_from_entity(seaql_migrations::Entity); @@ -131,8 +146,13 @@ pub trait MigratorTrait: Send { } /// Drop all tables from the database, then reapply all migrations - async fn fresh(db: &DbConn) -> Result<(), DbErr> { - Self::install(db).await?; + async fn fresh<'c, C>(db: C) -> Result<(), DbErr> + where + C: IntoSchemaManagerConnection<'c>, + { + let db = db.into_schema_manager_connection(); + + Self::install(&db).await?; let db_backend = db.get_database_backend(); // Temporarily disable the foreign key check @@ -149,7 +169,7 @@ pub trait MigratorTrait: Send { // Drop all foreign keys if db_backend == DbBackend::MySql { info!("Dropping all foreign keys"); - let stmt = query_mysql_foreign_keys(db); + let stmt = query_mysql_foreign_keys(&db); let rows = db.query_all(db_backend.build(&stmt)).await?; for row in rows.into_iter() { let constraint_name: String = row.try_get("", "CONSTRAINT_NAME")?; @@ -168,7 +188,7 @@ pub trait MigratorTrait: Send { } // Drop all tables - let stmt = query_tables(db); + let stmt = query_tables(&db); let rows = db.query_all(db_backend.build(&stmt)).await?; for row in rows.into_iter() { let table_name: String = row.try_get("", "table_name")?; @@ -184,7 +204,7 @@ pub trait MigratorTrait: Send { // Drop all types if db_backend == DbBackend::Postgres { info!("Dropping all types"); - let stmt = query_pg_types(db); + let stmt = query_pg_types(&db); let rows = db.query_all(db_backend.build(&stmt)).await?; for row in rows { let type_name: String = row.try_get("", "typname")?; @@ -212,18 +232,29 @@ pub trait MigratorTrait: Send { } /// Rollback all applied migrations, then reapply all migrations - async fn refresh(db: &DbConn) -> Result<(), DbErr> { - Self::down(db, None).await?; - Self::up(db, None).await + async fn refresh<'c, C>(db: C) -> Result<(), DbErr> + where + C: IntoSchemaManagerConnection<'c>, + { + let manager = SchemaManager::new(db); + Self::down_inner(&manager, None).await?; + Self::up_inner(&manager, None).await } /// Rollback all applied migrations - async fn reset(db: &DbConn) -> Result<(), DbErr> { - Self::down(db, None).await + async fn reset<'c, C>(db: C) -> Result<(), DbErr> + where + C: IntoSchemaManagerConnection<'c>, + { + let manager = SchemaManager::new(db); + Self::down_inner(&manager, None).await } /// Check the status of all migrations - async fn status(db: &DbConn) -> Result<(), DbErr> { + async fn status(db: &C) -> Result<(), DbErr> + where + C: ConnectionTrait, + { Self::install(db).await?; info!("Checking migration status"); @@ -236,9 +267,19 @@ pub trait MigratorTrait: Send { } /// Apply pending migrations - async fn up(db: &DbConn, mut steps: Option) -> Result<(), DbErr> { - Self::install(db).await?; + async fn up<'c, C>(db: C, steps: Option) -> Result<(), DbErr> + where + C: IntoSchemaManagerConnection<'c>, + { let manager = SchemaManager::new(db); + Self::up_inner(&manager, steps).await + } + + /// Apply pending migrations + async fn up_inner(manager: &SchemaManager, mut steps: Option) -> Result<(), DbErr> { + let db = manager.get_connection(); + + Self::install(db).await?; if let Some(steps) = steps { info!("Applying {} pending migrations", steps); @@ -258,7 +299,7 @@ pub trait MigratorTrait: Send { *steps -= 1; } info!("Applying migration '{}'", migration.name()); - migration.up(&manager).await?; + migration.up(manager).await?; info!("Migration '{}' has been applied", migration.name()); let now = SystemTime::now() .duration_since(SystemTime::UNIX_EPOCH) @@ -275,9 +316,19 @@ pub trait MigratorTrait: Send { } /// Rollback applied migrations - async fn down(db: &DbConn, mut steps: Option) -> Result<(), DbErr> { - Self::install(db).await?; + async fn down<'c, C>(db: C, steps: Option) -> Result<(), DbErr> + where + C: IntoSchemaManagerConnection<'c>, + { let manager = SchemaManager::new(db); + Self::down_inner(&manager, steps).await + } + + /// Rollback applied migrations + async fn down_inner(manager: &SchemaManager, mut steps: Option) -> Result<(), DbErr> { + let db = manager.get_connection(); + + Self::install(db).await?; if let Some(steps) = steps { info!("Rolling back {} applied migrations", steps); @@ -309,7 +360,10 @@ pub trait MigratorTrait: Send { } } -pub(crate) fn query_tables(db: &DbConn) -> SelectStatement { +pub(crate) fn query_tables(db: &C) -> SelectStatement +where + C: ConnectionTrait, +{ match db.get_database_backend() { DbBackend::MySql => MySql::query_tables(), DbBackend::Postgres => Postgres::query_tables(), @@ -317,7 +371,10 @@ pub(crate) fn query_tables(db: &DbConn) -> SelectStatement { } } -pub(crate) fn get_current_schema(db: &DbConn) -> SimpleExpr { +pub(crate) fn get_current_schema(db: &C) -> SimpleExpr +where + C: ConnectionTrait, +{ match db.get_database_backend() { DbBackend::MySql => MySql::get_current_schema(), DbBackend::Postgres => Postgres::get_current_schema(), @@ -338,7 +395,10 @@ enum InformationSchema { ConstraintType, } -fn query_mysql_foreign_keys(db: &DbConn) -> SelectStatement { +fn query_mysql_foreign_keys(db: &C) -> SelectStatement +where + C: ConnectionTrait, +{ let mut stmt = Query::select(); stmt.columns([ InformationSchema::TableName, @@ -380,7 +440,10 @@ enum PgNamespace { Nspname, } -fn query_pg_types(db: &DbConn) -> SelectStatement { +fn query_pg_types(db: &C) -> SelectStatement +where + C: ConnectionTrait, +{ let mut stmt = Query::select(); stmt.column(PgType::Typname) .from(PgType::Table) diff --git a/sea-orm-migration/src/prelude.rs b/sea-orm-migration/src/prelude.rs index fd78117e4..25bf9d5a6 100644 --- a/sea-orm-migration/src/prelude.rs +++ b/sea-orm-migration/src/prelude.rs @@ -1,12 +1,13 @@ #[cfg(feature = "cli")] -pub use super::cli; +pub use crate::cli; -pub use super::manager::SchemaManager; -pub use super::migrator::MigratorTrait; -pub use super::{MigrationName, MigrationTrait}; +pub use crate::{ + IntoSchemaManagerConnection, MigrationName, MigrationTrait, MigratorTrait, SchemaManager, + SchemaManagerConnection, +}; pub use async_trait; -pub use sea_orm; -pub use sea_orm::sea_query; -pub use sea_orm::sea_query::*; -pub use sea_orm::DbErr; -pub use sea_orm::DeriveMigrationName; +pub use sea_orm::{ + self, + sea_query::{self, *}, + DbErr, DeriveMigrationName, +}; diff --git a/sea-orm-migration/tests/main.rs b/sea-orm-migration/tests/main.rs index fe006c471..c1fa6eb40 100644 --- a/sea-orm-migration/tests/main.rs +++ b/sea-orm-migration/tests/main.rs @@ -1,7 +1,9 @@ mod migrator; use migrator::Migrator; -use sea_orm::{ConnectOptions, ConnectionTrait, Database, DbBackend, DbErr, Statement}; +use sea_orm::{ + ConnectOptions, ConnectionTrait, Database, DbBackend, DbErr, Statement, TransactionTrait, +}; use sea_orm_migration::prelude::*; #[async_std::test] @@ -67,85 +69,109 @@ async fn run_migration(url: &str, db_name: &str, schema: &str) -> Result<(), DbE } DbBackend::Sqlite => db, }; - let manager = SchemaManager::new(db); - println!("\nMigrator::status"); - Migrator::status(db).await?; + macro_rules! run_migration_tests { + ($db:expr, $manager:expr) => { + println!("\nMigrator::status"); + Migrator::status($db).await?; + + println!("\nMigrator::install"); + Migrator::install($db).await?; - println!("\nMigrator::install"); - Migrator::install(db).await?; + assert!($manager.has_table("seaql_migrations").await?); - assert!(manager.has_table("seaql_migrations").await?); + println!("\nMigrator::reset"); + Migrator::reset($db).await?; - println!("\nMigrator::reset"); - Migrator::reset(db).await?; + assert!(!$manager.has_table("cake").await?); + assert!(!$manager.has_table("fruit").await?); - assert!(!manager.has_table("cake").await?); - assert!(!manager.has_table("fruit").await?); + println!("\nMigrator::up"); + Migrator::up($db, Some(0)).await?; - println!("\nMigrator::up"); - Migrator::up(db, Some(0)).await?; + assert!(!$manager.has_table("cake").await?); + assert!(!$manager.has_table("fruit").await?); - assert!(!manager.has_table("cake").await?); - assert!(!manager.has_table("fruit").await?); + println!("\nMigrator::up"); + Migrator::up($db, Some(1)).await?; - println!("\nMigrator::up"); - Migrator::up(db, Some(1)).await?; + assert!($manager.has_table("cake").await?); + assert!(!$manager.has_table("fruit").await?); - assert!(manager.has_table("cake").await?); - assert!(!manager.has_table("fruit").await?); + println!("\nMigrator::down"); + Migrator::down($db, Some(0)).await?; - println!("\nMigrator::down"); - Migrator::down(db, Some(0)).await?; + assert!($manager.has_table("cake").await?); + assert!(!$manager.has_table("fruit").await?); - assert!(manager.has_table("cake").await?); - assert!(!manager.has_table("fruit").await?); + println!("\nMigrator::down"); + Migrator::down($db, Some(1)).await?; - println!("\nMigrator::down"); - Migrator::down(db, Some(1)).await?; + assert!(!$manager.has_table("cake").await?); + assert!(!$manager.has_table("fruit").await?); - assert!(!manager.has_table("cake").await?); - assert!(!manager.has_table("fruit").await?); + println!("\nMigrator::up"); + Migrator::up($db, None).await?; - println!("\nMigrator::up"); - Migrator::up(db, None).await?; + println!("\nMigrator::status"); + Migrator::status($db).await?; - println!("\nMigrator::status"); - Migrator::status(db).await?; + assert!($manager.has_table("cake").await?); + assert!($manager.has_table("fruit").await?); - assert!(manager.has_table("cake").await?); - assert!(manager.has_table("fruit").await?); + assert!($manager.has_column("cake", "name").await?); + assert!($manager.has_column("fruit", "cake_id").await?); - assert!(manager.has_column("cake", "name").await?); - assert!(manager.has_column("fruit", "cake_id").await?); + println!("\nMigrator::down"); + Migrator::down($db, None).await?; - println!("\nMigrator::down"); - Migrator::down(db, None).await?; + assert!($manager.has_table("seaql_migrations").await?); + assert!(!$manager.has_table("cake").await?); + assert!(!$manager.has_table("fruit").await?); - assert!(manager.has_table("seaql_migrations").await?); - assert!(!manager.has_table("cake").await?); - assert!(!manager.has_table("fruit").await?); + println!("\nMigrator::fresh"); + Migrator::fresh($db).await?; - println!("\nMigrator::fresh"); - Migrator::fresh(db).await?; + assert!($manager.has_table("cake").await?); + assert!($manager.has_table("fruit").await?); - assert!(manager.has_table("cake").await?); - assert!(manager.has_table("fruit").await?); + println!("\nMigrator::refresh"); + Migrator::refresh($db).await?; - println!("\nMigrator::refresh"); - Migrator::refresh(db).await?; + assert!($manager.has_table("cake").await?); + assert!($manager.has_table("fruit").await?); - assert!(manager.has_table("cake").await?); - assert!(manager.has_table("fruit").await?); + println!("\nMigrator::reset"); + Migrator::reset($db).await?; - println!("\nMigrator::reset"); - Migrator::reset(db).await?; + assert!(!$manager.has_table("cake").await?); + assert!(!$manager.has_table("fruit").await?); - assert!(!manager.has_table("cake").await?); - assert!(!manager.has_table("fruit").await?); + println!("\nMigrator::status"); + Migrator::status($db).await?; + }; + } - println!("\nMigrator::status"); - Migrator::status(db).await?; + // Run migration via `DatabaseConnection` + let manager = SchemaManager::new(db); + run_migration_tests!(db, manager); + + // Run migration via `DatabaseTransaction` + let transaction = db.begin().await?; + let manager = SchemaManager::new(&transaction); + run_migration_tests!(&transaction, manager); + transaction.commit().await?; + + // Run migration via transaction closure + db.transaction::<_, _, DbErr>(|transaction| { + Box::pin(async move { + let manager = SchemaManager::new(transaction); + run_migration_tests!(transaction, manager); + Ok(()) + }) + }) + .await + .unwrap(); Ok(()) } diff --git a/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs b/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs new file mode 100644 index 000000000..dc5f6a961 --- /dev/null +++ b/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs @@ -0,0 +1,57 @@ +use sea_orm_migration::prelude::*; +use sea_orm_migration::sea_orm::{entity::*, query::*}; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let db = manager.get_connection(); + + let transaction = db.begin().await?; + + cake::ActiveModel { + name: Set("Cheesecake".to_owned()), + ..Default::default() + } + .insert(&transaction) + .await?; + + transaction.commit().await?; + + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let db = manager.get_connection(); + + let transaction = db.begin().await?; + + cake::Entity::delete_many() + .filter(cake::Column::Name.eq("Cheesecake")) + .exec(&transaction) + .await?; + + transaction.commit().await?; + + Ok(()) + } +} + +mod cake { + use sea_orm_migration::sea_orm::entity::prelude::*; + + #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] + #[sea_orm(table_name = "cake")] + pub struct Model { + #[sea_orm(primary_key)] + pub id: i32, + pub name: String, + } + + #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] + pub enum Relation {} + + impl ActiveModelBehavior for ActiveModel {} +} diff --git a/sea-orm-migration/tests/migrator/mod.rs b/sea-orm-migration/tests/migrator/mod.rs index 042e5323a..7e7f68023 100644 --- a/sea-orm-migration/tests/migrator/mod.rs +++ b/sea-orm-migration/tests/migrator/mod.rs @@ -5,6 +5,7 @@ mod m20220118_000002_create_fruit_table; mod m20220118_000003_seed_cake_table; mod m20220118_000004_create_tea_enum; mod m20220923_000001_seed_cake_table; +mod m20230109_000001_seed_cake_table; pub struct Migrator; @@ -17,6 +18,7 @@ impl MigratorTrait for Migrator { Box::new(m20220118_000003_seed_cake_table::Migration), Box::new(m20220118_000004_create_tea_enum::Migration), Box::new(m20220923_000001_seed_cake_table::Migration), + Box::new(m20230109_000001_seed_cake_table::Migration), ] } } From 1a3be36f6cd8dc4eeb6b34500de5a2a89c81d1a9 Mon Sep 17 00:00:00 2001 From: Billy Chan Date: Fri, 20 Jan 2023 00:27:02 +0800 Subject: [PATCH 2/9] refactor --- sea-orm-migration/src/manager.rs | 4 +- sea-orm-migration/src/migrator.rs | 152 +++++++++++++++--------------- 2 files changed, 79 insertions(+), 77 deletions(-) diff --git a/sea-orm-migration/src/manager.rs b/sea-orm-migration/src/manager.rs index 88b1235d1..f70c9212a 100644 --- a/sea-orm-migration/src/manager.rs +++ b/sea-orm-migration/src/manager.rs @@ -5,9 +5,7 @@ use sea_orm::sea_query::{ TableAlterStatement, TableCreateStatement, TableDropStatement, TableRenameStatement, TableTruncateStatement, }; -use sea_orm::{ - ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbBackend, DbErr, StatementBuilder, -}; +use sea_orm::{ConnectionTrait, DbBackend, DbErr, StatementBuilder}; use sea_schema::{mysql::MySql, postgres::Postgres, probe::SchemaProbe, sqlite::Sqlite}; /// Helper struct for writing migration scripts in migration file diff --git a/sea-orm-migration/src/migrator.rs b/sea-orm-migration/src/migrator.rs index cf54aa23f..e71d07aba 100644 --- a/sea-orm-migration/src/migrator.rs +++ b/sea-orm-migration/src/migrator.rs @@ -8,8 +8,8 @@ use sea_orm::sea_query::{ SelectStatement, SimpleExpr, Table, }; use sea_orm::{ - ActiveModelTrait, ActiveValue, ColumnTrait, Condition, ConnectionTrait, DbBackend, DbConn, - DbErr, EntityTrait, QueryFilter, QueryOrder, Schema, Statement, + ActiveModelTrait, ActiveValue, ColumnTrait, Condition, ConnectionTrait, DbBackend, DbErr, + EntityTrait, QueryFilter, QueryOrder, Schema, Statement, }; use sea_schema::{mysql::MySql, postgres::Postgres, probe::SchemaProbe, sqlite::Sqlite}; @@ -237,8 +237,8 @@ pub trait MigratorTrait: Send { C: IntoSchemaManagerConnection<'c>, { let manager = SchemaManager::new(db); - Self::down_inner(&manager, None).await?; - Self::up_inner(&manager, None).await + down_internal::(&manager, None).await?; + up_internal::(&manager, None).await } /// Rollback all applied migrations @@ -247,7 +247,7 @@ pub trait MigratorTrait: Send { C: IntoSchemaManagerConnection<'c>, { let manager = SchemaManager::new(db); - Self::down_inner(&manager, None).await + down_internal::(&manager, None).await } /// Check the status of all migrations @@ -272,47 +272,7 @@ pub trait MigratorTrait: Send { C: IntoSchemaManagerConnection<'c>, { let manager = SchemaManager::new(db); - Self::up_inner(&manager, steps).await - } - - /// Apply pending migrations - async fn up_inner(manager: &SchemaManager, mut steps: Option) -> Result<(), DbErr> { - let db = manager.get_connection(); - - Self::install(db).await?; - - if let Some(steps) = steps { - info!("Applying {} pending migrations", steps); - } else { - info!("Applying all pending migrations"); - } - - let migrations = Self::get_pending_migrations(db).await?.into_iter(); - if migrations.len() == 0 { - info!("No pending migrations"); - } - for Migration { migration, .. } in migrations { - if let Some(steps) = steps.as_mut() { - if steps == &0 { - break; - } - *steps -= 1; - } - info!("Applying migration '{}'", migration.name()); - migration.up(manager).await?; - info!("Migration '{}' has been applied", migration.name()); - let now = SystemTime::now() - .duration_since(SystemTime::UNIX_EPOCH) - .expect("SystemTime before UNIX EPOCH!"); - seaql_migrations::ActiveModel { - version: ActiveValue::Set(migration.name().to_owned()), - applied_at: ActiveValue::Set(now.as_secs() as i64), - } - .insert(db) - .await?; - } - - Ok(()) + up_internal::(&manager, steps).await } /// Rollback applied migrations @@ -321,43 +281,87 @@ pub trait MigratorTrait: Send { C: IntoSchemaManagerConnection<'c>, { let manager = SchemaManager::new(db); - Self::down_inner(&manager, steps).await + down_internal::(&manager, steps).await } +} - /// Rollback applied migrations - async fn down_inner(manager: &SchemaManager, mut steps: Option) -> Result<(), DbErr> { - let db = manager.get_connection(); +async fn up_internal(manager: &SchemaManager<'_>, mut steps: Option) -> Result<(), DbErr> +where + M: MigratorTrait + ?Sized, +{ + let db = manager.get_connection(); - Self::install(db).await?; + M::install(db).await?; - if let Some(steps) = steps { - info!("Rolling back {} applied migrations", steps); - } else { - info!("Rolling back all applied migrations"); - } + if let Some(steps) = steps { + info!("Applying {} pending migrations", steps); + } else { + info!("Applying all pending migrations"); + } - let migrations = Self::get_applied_migrations(db).await?.into_iter().rev(); - if migrations.len() == 0 { - info!("No applied migrations"); - } - for Migration { migration, .. } in migrations { - if let Some(steps) = steps.as_mut() { - if steps == &0 { - break; - } - *steps -= 1; + let migrations = M::get_pending_migrations(db).await?.into_iter(); + if migrations.len() == 0 { + info!("No pending migrations"); + } + for Migration { migration, .. } in migrations { + if let Some(steps) = steps.as_mut() { + if steps == &0 { + break; } - info!("Rolling back migration '{}'", migration.name()); - migration.down(&manager).await?; - info!("Migration '{}' has been rollbacked", migration.name()); - seaql_migrations::Entity::delete_many() - .filter(seaql_migrations::Column::Version.eq(migration.name())) - .exec(db) - .await?; + *steps -= 1; + } + info!("Applying migration '{}'", migration.name()); + migration.up(manager).await?; + info!("Migration '{}' has been applied", migration.name()); + let now = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .expect("SystemTime before UNIX EPOCH!"); + seaql_migrations::ActiveModel { + version: ActiveValue::Set(migration.name().to_owned()), + applied_at: ActiveValue::Set(now.as_secs() as i64), } + .insert(db) + .await?; + } - Ok(()) + Ok(()) +} + +async fn down_internal(manager: &SchemaManager<'_>, mut steps: Option) -> Result<(), DbErr> +where + M: MigratorTrait + ?Sized, +{ + let db = manager.get_connection(); + + M::install(db).await?; + + if let Some(steps) = steps { + info!("Rolling back {} applied migrations", steps); + } else { + info!("Rolling back all applied migrations"); + } + + let migrations = M::get_applied_migrations(db).await?.into_iter().rev(); + if migrations.len() == 0 { + info!("No applied migrations"); } + for Migration { migration, .. } in migrations { + if let Some(steps) = steps.as_mut() { + if steps == &0 { + break; + } + *steps -= 1; + } + info!("Rolling back migration '{}'", migration.name()); + migration.down(&manager).await?; + info!("Migration '{}' has been rollbacked", migration.name()); + seaql_migrations::Entity::delete_many() + .filter(seaql_migrations::Column::Version.eq(migration.name())) + .exec(db) + .await?; + } + + Ok(()) } pub(crate) fn query_tables(db: &C) -> SelectStatement From c3dcb8b7fd586c9770fc0007de8d7c9203e3c569 Mon Sep 17 00:00:00 2001 From: Billy Chan Date: Fri, 20 Jan 2023 12:18:34 +0800 Subject: [PATCH 3/9] clippy --- sea-orm-migration/src/migrator.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sea-orm-migration/src/migrator.rs b/sea-orm-migration/src/migrator.rs index e71d07aba..8c2ff3015 100644 --- a/sea-orm-migration/src/migrator.rs +++ b/sea-orm-migration/src/migrator.rs @@ -353,7 +353,7 @@ where *steps -= 1; } info!("Rolling back migration '{}'", migration.name()); - migration.down(&manager).await?; + migration.down(manager).await?; info!("Migration '{}' has been rollbacked", migration.name()); seaql_migrations::Entity::delete_many() .filter(seaql_migrations::Column::Version.eq(migration.name())) From 18ae130121c8891da6af44e6b8ca4e1927930fa8 Mon Sep 17 00:00:00 2001 From: Billy Chan Date: Fri, 20 Jan 2023 12:30:31 +0800 Subject: [PATCH 4/9] visibility --- sea-orm-migration/src/migrator.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sea-orm-migration/src/migrator.rs b/sea-orm-migration/src/migrator.rs index 8c2ff3015..9209bdeb0 100644 --- a/sea-orm-migration/src/migrator.rs +++ b/sea-orm-migration/src/migrator.rs @@ -364,7 +364,7 @@ where Ok(()) } -pub(crate) fn query_tables(db: &C) -> SelectStatement +fn query_tables(db: &C) -> SelectStatement where C: ConnectionTrait, { @@ -375,7 +375,7 @@ where } } -pub(crate) fn get_current_schema(db: &C) -> SimpleExpr +fn get_current_schema(db: &C) -> SimpleExpr where C: ConnectionTrait, { From 2db29c3c27f338ddcfac5c39800bd7aee3210535 Mon Sep 17 00:00:00 2001 From: Billy Chan Date: Fri, 20 Jan 2023 16:40:46 +0800 Subject: [PATCH 5/9] Execute migration in transaction when it's a Postgres connection --- sea-orm-migration/src/migrator.rs | 268 +++++++++++------- sea-orm-migration/tests/main.rs | 159 +++++------ .../m20220923_000001_seed_cake_table.rs | 2 - .../m20230109_000001_seed_cake_table.rs | 6 + 4 files changed, 246 insertions(+), 189 deletions(-) diff --git a/sea-orm-migration/src/migrator.rs b/sea-orm-migration/src/migrator.rs index 9209bdeb0..51c1f101d 100644 --- a/sea-orm-migration/src/migrator.rs +++ b/sea-orm-migration/src/migrator.rs @@ -1,5 +1,7 @@ +use futures::Future; use std::collections::HashSet; use std::fmt::Display; +use std::pin::Pin; use std::time::SystemTime; use tracing::info; @@ -9,7 +11,7 @@ use sea_orm::sea_query::{ }; use sea_orm::{ ActiveModelTrait, ActiveValue, ColumnTrait, Condition, ConnectionTrait, DbBackend, DbErr, - EntityTrait, QueryFilter, QueryOrder, Schema, Statement, + EntityTrait, QueryFilter, QueryOrder, Schema, Statement, TransactionTrait, }; use sea_schema::{mysql::MySql, postgres::Postgres, probe::SchemaProbe, sqlite::Sqlite}; @@ -145,147 +147,197 @@ pub trait MigratorTrait: Send { db.execute(builder.build(&stmt)).await.map(|_| ()) } - /// Drop all tables from the database, then reapply all migrations - async fn fresh<'c, C>(db: C) -> Result<(), DbErr> + /// Check the status of all migrations + async fn status(db: &C) -> Result<(), DbErr> where - C: IntoSchemaManagerConnection<'c>, + C: ConnectionTrait, { - let db = db.into_schema_manager_connection(); - - Self::install(&db).await?; - let db_backend = db.get_database_backend(); - - // Temporarily disable the foreign key check - if db_backend == DbBackend::Sqlite { - info!("Disabling foreign key check"); - db.execute(Statement::from_string( - db_backend, - "PRAGMA foreign_keys = OFF".to_owned(), - )) - .await?; - info!("Foreign key check disabled"); - } - - // Drop all foreign keys - if db_backend == DbBackend::MySql { - info!("Dropping all foreign keys"); - let stmt = query_mysql_foreign_keys(&db); - let rows = db.query_all(db_backend.build(&stmt)).await?; - for row in rows.into_iter() { - let constraint_name: String = row.try_get("", "CONSTRAINT_NAME")?; - let table_name: String = row.try_get("", "TABLE_NAME")?; - info!( - "Dropping foreign key '{}' from table '{}'", - constraint_name, table_name - ); - let mut stmt = ForeignKey::drop(); - stmt.table(Alias::new(table_name.as_str())) - .name(constraint_name.as_str()); - db.execute(db_backend.build(&stmt)).await?; - info!("Foreign key '{}' has been dropped", constraint_name); - } - info!("All foreign keys dropped"); - } - - // Drop all tables - let stmt = query_tables(&db); - let rows = db.query_all(db_backend.build(&stmt)).await?; - for row in rows.into_iter() { - let table_name: String = row.try_get("", "table_name")?; - info!("Dropping table '{}'", table_name); - let mut stmt = Table::drop(); - stmt.table(Alias::new(table_name.as_str())) - .if_exists() - .cascade(); - db.execute(db_backend.build(&stmt)).await?; - info!("Table '{}' has been dropped", table_name); - } + Self::install(db).await?; - // Drop all types - if db_backend == DbBackend::Postgres { - info!("Dropping all types"); - let stmt = query_pg_types(&db); - let rows = db.query_all(db_backend.build(&stmt)).await?; - for row in rows { - let type_name: String = row.try_get("", "typname")?; - info!("Dropping type '{}'", type_name); - let mut stmt = Type::drop(); - stmt.name(Alias::new(&type_name as &str)); - db.execute(db_backend.build(&stmt)).await?; - info!("Type '{}' has been dropped", type_name); - } - } + info!("Checking migration status"); - // Restore the foreign key check - if db_backend == DbBackend::Sqlite { - info!("Restoring foreign key check"); - db.execute(Statement::from_string( - db_backend, - "PRAGMA foreign_keys = ON".to_owned(), - )) - .await?; - info!("Foreign key check restored"); + for Migration { migration, status } in Self::get_migration_with_status(db).await? { + info!("Migration '{}'... {}", migration.name(), status); } - // Reapply all migrations - Self::up(db, None).await + Ok(()) } - /// Rollback all applied migrations, then reapply all migrations - async fn refresh<'c, C>(db: C) -> Result<(), DbErr> + /// Drop all tables from the database, then reapply all migrations + async fn fresh<'a, 'c, C>(db: C) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, + Self: 'a, { - let manager = SchemaManager::new(db); - down_internal::(&manager, None).await?; - up_internal::(&manager, None).await + exec_with_connection::<'a, '_, _, _, Self>(db, move |manager| { + Box::pin(async move { exec_fresh::(manager).await }) + }) + .await } - /// Rollback all applied migrations - async fn reset<'c, C>(db: C) -> Result<(), DbErr> + /// Rollback all applied migrations, then reapply all migrations + async fn refresh<'a, 'c, C>(db: C) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, + Self: 'a, { - let manager = SchemaManager::new(db); - down_internal::(&manager, None).await + exec_with_connection::<'a, '_, _, _, Self>(db, move |manager| { + Box::pin(async move { + exec_down::(manager, None).await?; + exec_up::(manager, None).await + }) + }) + .await } - /// Check the status of all migrations - async fn status(db: &C) -> Result<(), DbErr> + /// Rollback all applied migrations + async fn reset<'a, 'c, C>(db: C) -> Result<(), DbErr> where - C: ConnectionTrait, + C: IntoSchemaManagerConnection<'c>, + Self: 'a, { - Self::install(db).await?; - - info!("Checking migration status"); - - for Migration { migration, status } in Self::get_migration_with_status(db).await? { - info!("Migration '{}'... {}", migration.name(), status); - } - - Ok(()) + exec_with_connection::<'a, '_, _, _, Self>(db, move |manager| { + Box::pin(async move { exec_down::(manager, None).await }) + }) + .await } /// Apply pending migrations - async fn up<'c, C>(db: C, steps: Option) -> Result<(), DbErr> + async fn up<'a, 'c, C>(db: C, steps: Option) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, + Self: 'a, { - let manager = SchemaManager::new(db); - up_internal::(&manager, steps).await + exec_with_connection::<'a, '_, _, _, Self>(db, move |manager| { + Box::pin(async move { exec_up::(manager, steps).await }) + }) + .await } /// Rollback applied migrations - async fn down<'c, C>(db: C, steps: Option) -> Result<(), DbErr> + async fn down<'a, 'c, C>(db: C, steps: Option) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, + Self: 'a, { - let manager = SchemaManager::new(db); - down_internal::(&manager, steps).await + exec_with_connection::<'a, '_, _, _, Self>(db, move |manager| { + Box::pin(async move { exec_down::(manager, steps).await }) + }) + .await + } +} + +async fn exec_with_connection<'a, 'c, C, F, M>(db: C, f: F) -> Result<(), DbErr> +where + C: IntoSchemaManagerConnection<'c>, + F: for<'b> Fn( + &'b SchemaManager<'_>, + ) -> Pin> + Send + 'b>> + + 'a, + M: MigratorTrait + ?Sized + 'a, +{ + let db = db.into_schema_manager_connection(); + + match db.get_database_backend() { + DbBackend::Postgres => { + let transaction = db.begin().await?; + let manager = SchemaManager::new(&transaction); + f(&manager).await?; + transaction.commit().await + } + DbBackend::MySql | DbBackend::Sqlite => { + let manager = SchemaManager::new(db); + f(&manager).await + } + } +} + +async fn exec_fresh(manager: &SchemaManager<'_>) -> Result<(), DbErr> +where + M: MigratorTrait + ?Sized, +{ + let db = manager.get_connection(); + + M::install(db).await?; + let db_backend = db.get_database_backend(); + + // Temporarily disable the foreign key check + if db_backend == DbBackend::Sqlite { + info!("Disabling foreign key check"); + db.execute(Statement::from_string( + db_backend, + "PRAGMA foreign_keys = OFF".to_owned(), + )) + .await?; + info!("Foreign key check disabled"); + } + + // Drop all foreign keys + if db_backend == DbBackend::MySql { + info!("Dropping all foreign keys"); + let stmt = query_mysql_foreign_keys(db); + let rows = db.query_all(db_backend.build(&stmt)).await?; + for row in rows.into_iter() { + let constraint_name: String = row.try_get("", "CONSTRAINT_NAME")?; + let table_name: String = row.try_get("", "TABLE_NAME")?; + info!( + "Dropping foreign key '{}' from table '{}'", + constraint_name, table_name + ); + let mut stmt = ForeignKey::drop(); + stmt.table(Alias::new(table_name.as_str())) + .name(constraint_name.as_str()); + db.execute(db_backend.build(&stmt)).await?; + info!("Foreign key '{}' has been dropped", constraint_name); + } + info!("All foreign keys dropped"); + } + + // Drop all tables + let stmt = query_tables(db); + let rows = db.query_all(db_backend.build(&stmt)).await?; + for row in rows.into_iter() { + let table_name: String = row.try_get("", "table_name")?; + info!("Dropping table '{}'", table_name); + let mut stmt = Table::drop(); + stmt.table(Alias::new(table_name.as_str())) + .if_exists() + .cascade(); + db.execute(db_backend.build(&stmt)).await?; + info!("Table '{}' has been dropped", table_name); + } + + // Drop all types + if db_backend == DbBackend::Postgres { + info!("Dropping all types"); + let stmt = query_pg_types(db); + let rows = db.query_all(db_backend.build(&stmt)).await?; + for row in rows { + let type_name: String = row.try_get("", "typname")?; + info!("Dropping type '{}'", type_name); + let mut stmt = Type::drop(); + stmt.name(Alias::new(&type_name as &str)); + db.execute(db_backend.build(&stmt)).await?; + info!("Type '{}' has been dropped", type_name); + } } + + // Restore the foreign key check + if db_backend == DbBackend::Sqlite { + info!("Restoring foreign key check"); + db.execute(Statement::from_string( + db_backend, + "PRAGMA foreign_keys = ON".to_owned(), + )) + .await?; + info!("Foreign key check restored"); + } + + // Reapply all migrations + exec_up::(manager, None).await } -async fn up_internal(manager: &SchemaManager<'_>, mut steps: Option) -> Result<(), DbErr> +async fn exec_up(manager: &SchemaManager<'_>, mut steps: Option) -> Result<(), DbErr> where M: MigratorTrait + ?Sized, { @@ -327,7 +379,7 @@ where Ok(()) } -async fn down_internal(manager: &SchemaManager<'_>, mut steps: Option) -> Result<(), DbErr> +async fn exec_down(manager: &SchemaManager<'_>, mut steps: Option) -> Result<(), DbErr> where M: MigratorTrait + ?Sized, { diff --git a/sea-orm-migration/tests/main.rs b/sea-orm-migration/tests/main.rs index c1fa6eb40..3588a097c 100644 --- a/sea-orm-migration/tests/main.rs +++ b/sea-orm-migration/tests/main.rs @@ -1,9 +1,7 @@ mod migrator; use migrator::Migrator; -use sea_orm::{ - ConnectOptions, ConnectionTrait, Database, DbBackend, DbErr, Statement, TransactionTrait, -}; +use sea_orm::{ConnectOptions, ConnectionTrait, Database, DbBackend, DbErr, Statement}; use sea_orm_migration::prelude::*; #[async_std::test] @@ -69,109 +67,112 @@ async fn run_migration(url: &str, db_name: &str, schema: &str) -> Result<(), DbE } DbBackend::Sqlite => db, }; + let manager = SchemaManager::new(db); - macro_rules! run_migration_tests { - ($db:expr, $manager:expr) => { - println!("\nMigrator::status"); - Migrator::status($db).await?; + println!("\nMigrator::status"); + Migrator::status(db).await?; - println!("\nMigrator::install"); - Migrator::install($db).await?; + println!("\nMigrator::install"); + Migrator::install(db).await?; - assert!($manager.has_table("seaql_migrations").await?); + assert!(manager.has_table("seaql_migrations").await?); - println!("\nMigrator::reset"); - Migrator::reset($db).await?; + println!("\nMigrator::reset"); + Migrator::reset(db).await?; - assert!(!$manager.has_table("cake").await?); - assert!(!$manager.has_table("fruit").await?); + assert!(!manager.has_table("cake").await?); + assert!(!manager.has_table("fruit").await?); - println!("\nMigrator::up"); - Migrator::up($db, Some(0)).await?; + println!("\nMigrator::up"); + Migrator::up(db, Some(0)).await?; - assert!(!$manager.has_table("cake").await?); - assert!(!$manager.has_table("fruit").await?); + assert!(!manager.has_table("cake").await?); + assert!(!manager.has_table("fruit").await?); - println!("\nMigrator::up"); - Migrator::up($db, Some(1)).await?; + println!("\nMigrator::up"); + Migrator::up(db, Some(1)).await?; - assert!($manager.has_table("cake").await?); - assert!(!$manager.has_table("fruit").await?); + assert!(manager.has_table("cake").await?); + assert!(!manager.has_table("fruit").await?); - println!("\nMigrator::down"); - Migrator::down($db, Some(0)).await?; + println!("\nMigrator::down"); + Migrator::down(db, Some(0)).await?; - assert!($manager.has_table("cake").await?); - assert!(!$manager.has_table("fruit").await?); + assert!(manager.has_table("cake").await?); + assert!(!manager.has_table("fruit").await?); - println!("\nMigrator::down"); - Migrator::down($db, Some(1)).await?; + println!("\nMigrator::down"); + Migrator::down(db, Some(1)).await?; - assert!(!$manager.has_table("cake").await?); - assert!(!$manager.has_table("fruit").await?); + assert!(!manager.has_table("cake").await?); + assert!(!manager.has_table("fruit").await?); - println!("\nMigrator::up"); - Migrator::up($db, None).await?; + // Tests rolling back changes when running migration on Postgres + if matches!(db.get_database_backend(), DbBackend::Postgres) { + println!("\nRoll back changes when encounter errors"); - println!("\nMigrator::status"); - Migrator::status($db).await?; + // Set a flag to throw error inside `m20230109_000001_seed_cake_table.rs` + std::env::set_var("ABOARD_MIGRATION", "YES"); - assert!($manager.has_table("cake").await?); - assert!($manager.has_table("fruit").await?); + // Should throw an error + println!("\nMigrator::up"); + assert_eq!( + Migrator::up(db, None).await, + Err(DbErr::Migration( + "Aboard migration and rollback changes".into() + )) + ); - assert!($manager.has_column("cake", "name").await?); - assert!($manager.has_column("fruit", "cake_id").await?); + println!("\nMigrator::status"); + Migrator::status(db).await?; - println!("\nMigrator::down"); - Migrator::down($db, None).await?; + // Check migrations have been rolled back + assert!(!manager.has_table("cake").await?); + assert!(!manager.has_table("fruit").await?); - assert!($manager.has_table("seaql_migrations").await?); - assert!(!$manager.has_table("cake").await?); - assert!(!$manager.has_table("fruit").await?); + // Unset the flag + std::env::remove_var("ABOARD_MIGRATION"); + } - println!("\nMigrator::fresh"); - Migrator::fresh($db).await?; + println!("\nMigrator::up"); + Migrator::up(db, None).await?; - assert!($manager.has_table("cake").await?); - assert!($manager.has_table("fruit").await?); + println!("\nMigrator::status"); + Migrator::status(db).await?; - println!("\nMigrator::refresh"); - Migrator::refresh($db).await?; + assert!(manager.has_table("cake").await?); + assert!(manager.has_table("fruit").await?); - assert!($manager.has_table("cake").await?); - assert!($manager.has_table("fruit").await?); + assert!(manager.has_column("cake", "name").await?); + assert!(manager.has_column("fruit", "cake_id").await?); - println!("\nMigrator::reset"); - Migrator::reset($db).await?; + println!("\nMigrator::down"); + Migrator::down(db, None).await?; - assert!(!$manager.has_table("cake").await?); - assert!(!$manager.has_table("fruit").await?); + assert!(manager.has_table("seaql_migrations").await?); + assert!(!manager.has_table("cake").await?); + assert!(!manager.has_table("fruit").await?); - println!("\nMigrator::status"); - Migrator::status($db).await?; - }; - } + println!("\nMigrator::fresh"); + Migrator::fresh(db).await?; - // Run migration via `DatabaseConnection` - let manager = SchemaManager::new(db); - run_migration_tests!(db, manager); - - // Run migration via `DatabaseTransaction` - let transaction = db.begin().await?; - let manager = SchemaManager::new(&transaction); - run_migration_tests!(&transaction, manager); - transaction.commit().await?; - - // Run migration via transaction closure - db.transaction::<_, _, DbErr>(|transaction| { - Box::pin(async move { - let manager = SchemaManager::new(transaction); - run_migration_tests!(transaction, manager); - Ok(()) - }) - }) - .await - .unwrap(); + assert!(manager.has_table("cake").await?); + assert!(manager.has_table("fruit").await?); + + println!("\nMigrator::refresh"); + Migrator::refresh(db).await?; + + assert!(manager.has_table("cake").await?); + assert!(manager.has_table("fruit").await?); + + println!("\nMigrator::reset"); + Migrator::reset(db).await?; + + assert!(!manager.has_table("cake").await?); + assert!(!manager.has_table("fruit").await?); + + println!("\nMigrator::status"); + Migrator::status(db).await?; Ok(()) } diff --git a/sea-orm-migration/tests/migrator/m20220923_000001_seed_cake_table.rs b/sea-orm-migration/tests/migrator/m20220923_000001_seed_cake_table.rs index ece73e1a7..16c019bf8 100644 --- a/sea-orm-migration/tests/migrator/m20220923_000001_seed_cake_table.rs +++ b/sea-orm-migration/tests/migrator/m20220923_000001_seed_cake_table.rs @@ -1,5 +1,4 @@ use sea_orm_migration::prelude::*; -use sea_orm_migration::sea_orm::{entity::*, query::*}; #[derive(DeriveMigrationName)] pub struct Migration; @@ -34,6 +33,5 @@ impl MigrationTrait for Migration { #[derive(Iden)] pub enum Cake { Table, - Id, Name, } diff --git a/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs b/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs index dc5f6a961..7f8498252 100644 --- a/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs +++ b/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs @@ -20,6 +20,12 @@ impl MigrationTrait for Migration { transaction.commit().await?; + if std::env::var_os("ABOARD_MIGRATION").eq(&Some("YES".into())) { + return Err(DbErr::Migration( + "Aboard migration and rollback changes".into(), + )); + } + Ok(()) } From 6382432206149e32eedfe6c0610099f0c03eb2a6 Mon Sep 17 00:00:00 2001 From: Billy Chan Date: Fri, 20 Jan 2023 16:44:35 +0800 Subject: [PATCH 6/9] Testing rolling back on migrate up and down --- sea-orm-migration/tests/main.rs | 29 ++++++++++++++++++- .../m20220118_000001_create_cake_table.rs | 10 ++++++- 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/sea-orm-migration/tests/main.rs b/sea-orm-migration/tests/main.rs index 3588a097c..2835dfe02 100644 --- a/sea-orm-migration/tests/main.rs +++ b/sea-orm-migration/tests/main.rs @@ -107,7 +107,7 @@ async fn run_migration(url: &str, db_name: &str, schema: &str) -> Result<(), DbE assert!(!manager.has_table("cake").await?); assert!(!manager.has_table("fruit").await?); - // Tests rolling back changes when running migration on Postgres + // Tests rolling back changes of "migrate up" when running migration on Postgres if matches!(db.get_database_backend(), DbBackend::Postgres) { println!("\nRoll back changes when encounter errors"); @@ -146,6 +146,33 @@ async fn run_migration(url: &str, db_name: &str, schema: &str) -> Result<(), DbE assert!(manager.has_column("cake", "name").await?); assert!(manager.has_column("fruit", "cake_id").await?); + // Tests rolling back changes of "migrate down" when running migration on Postgres + if matches!(db.get_database_backend(), DbBackend::Postgres) { + println!("\nRoll back changes when encounter errors"); + + // Set a flag to throw error inside `m20230109_000001_seed_cake_table.rs` + std::env::set_var("ABOARD_MIGRATION", "YES"); + + // Should throw an error + println!("\nMigrator::down"); + assert_eq!( + Migrator::down(db, None).await, + Err(DbErr::Migration( + "Aboard migration and rollback changes".into() + )) + ); + + println!("\nMigrator::status"); + Migrator::status(db).await?; + + // Check migrations have been rolled back + assert!(manager.has_table("cake").await?); + assert!(manager.has_table("fruit").await?); + + // Unset the flag + std::env::remove_var("ABOARD_MIGRATION"); + } + println!("\nMigrator::down"); Migrator::down(db, None).await?; diff --git a/sea-orm-migration/tests/migrator/m20220118_000001_create_cake_table.rs b/sea-orm-migration/tests/migrator/m20220118_000001_create_cake_table.rs index 3ad76eab4..28bc3e5bf 100644 --- a/sea-orm-migration/tests/migrator/m20220118_000001_create_cake_table.rs +++ b/sea-orm-migration/tests/migrator/m20220118_000001_create_cake_table.rs @@ -26,7 +26,15 @@ impl MigrationTrait for Migration { async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { manager .drop_table(Table::drop().table(Cake::Table).to_owned()) - .await + .await?; + + if std::env::var_os("ABOARD_MIGRATION").eq(&Some("YES".into())) { + return Err(DbErr::Migration( + "Aboard migration and rollback changes".into(), + )); + } + + Ok(()) } } From d07b23e5e91f0be8d5bf56be97ef109cdee378a6 Mon Sep 17 00:00:00 2001 From: Billy Chan Date: Fri, 20 Jan 2023 16:50:41 +0800 Subject: [PATCH 7/9] lifetimes elision --- sea-orm-migration/src/migrator.rs | 34 +++++++++++++------------------ 1 file changed, 14 insertions(+), 20 deletions(-) diff --git a/sea-orm-migration/src/migrator.rs b/sea-orm-migration/src/migrator.rs index 51c1f101d..8349bb59b 100644 --- a/sea-orm-migration/src/migrator.rs +++ b/sea-orm-migration/src/migrator.rs @@ -164,24 +164,22 @@ pub trait MigratorTrait: Send { } /// Drop all tables from the database, then reapply all migrations - async fn fresh<'a, 'c, C>(db: C) -> Result<(), DbErr> + async fn fresh<'c, C>(db: C) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, - Self: 'a, { - exec_with_connection::<'a, '_, _, _, Self>(db, move |manager| { + exec_with_connection::<'_, _, _, Self>(db, move |manager| { Box::pin(async move { exec_fresh::(manager).await }) }) .await } /// Rollback all applied migrations, then reapply all migrations - async fn refresh<'a, 'c, C>(db: C) -> Result<(), DbErr> + async fn refresh<'c, C>(db: C) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, - Self: 'a, { - exec_with_connection::<'a, '_, _, _, Self>(db, move |manager| { + exec_with_connection::<'_, _, _, Self>(db, move |manager| { Box::pin(async move { exec_down::(manager, None).await?; exec_up::(manager, None).await @@ -191,50 +189,46 @@ pub trait MigratorTrait: Send { } /// Rollback all applied migrations - async fn reset<'a, 'c, C>(db: C) -> Result<(), DbErr> + async fn reset<'c, C>(db: C) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, - Self: 'a, { - exec_with_connection::<'a, '_, _, _, Self>(db, move |manager| { + exec_with_connection::<'_, _, _, Self>(db, move |manager| { Box::pin(async move { exec_down::(manager, None).await }) }) .await } /// Apply pending migrations - async fn up<'a, 'c, C>(db: C, steps: Option) -> Result<(), DbErr> + async fn up<'c, C>(db: C, steps: Option) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, - Self: 'a, { - exec_with_connection::<'a, '_, _, _, Self>(db, move |manager| { + exec_with_connection::<'_, _, _, Self>(db, move |manager| { Box::pin(async move { exec_up::(manager, steps).await }) }) .await } /// Rollback applied migrations - async fn down<'a, 'c, C>(db: C, steps: Option) -> Result<(), DbErr> + async fn down<'c, C>(db: C, steps: Option) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, - Self: 'a, { - exec_with_connection::<'a, '_, _, _, Self>(db, move |manager| { + exec_with_connection::<'_, _, _, Self>(db, move |manager| { Box::pin(async move { exec_down::(manager, steps).await }) }) .await } } -async fn exec_with_connection<'a, 'c, C, F, M>(db: C, f: F) -> Result<(), DbErr> +async fn exec_with_connection<'c, C, F, M>(db: C, f: F) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, F: for<'b> Fn( - &'b SchemaManager<'_>, - ) -> Pin> + Send + 'b>> - + 'a, - M: MigratorTrait + ?Sized + 'a, + &'b SchemaManager<'_>, + ) -> Pin> + Send + 'b>>, + M: MigratorTrait + ?Sized, { let db = db.into_schema_manager_connection(); From 8d26dfe1f83ad9cb2a50f4aae3031b325bebc716 Mon Sep 17 00:00:00 2001 From: Billy Chan Date: Mon, 30 Jan 2023 12:45:47 +0800 Subject: [PATCH 8/9] typo --- sea-orm-migration/tests/main.rs | 8 ++++---- .../tests/migrator/m20220118_000001_create_cake_table.rs | 2 +- .../tests/migrator/m20230109_000001_seed_cake_table.rs | 2 +- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/sea-orm-migration/tests/main.rs b/sea-orm-migration/tests/main.rs index 2835dfe02..fcee6f9bf 100644 --- a/sea-orm-migration/tests/main.rs +++ b/sea-orm-migration/tests/main.rs @@ -112,7 +112,7 @@ async fn run_migration(url: &str, db_name: &str, schema: &str) -> Result<(), DbE println!("\nRoll back changes when encounter errors"); // Set a flag to throw error inside `m20230109_000001_seed_cake_table.rs` - std::env::set_var("ABOARD_MIGRATION", "YES"); + std::env::set_var("ABORT_MIGRATION", "YES"); // Should throw an error println!("\nMigrator::up"); @@ -131,7 +131,7 @@ async fn run_migration(url: &str, db_name: &str, schema: &str) -> Result<(), DbE assert!(!manager.has_table("fruit").await?); // Unset the flag - std::env::remove_var("ABOARD_MIGRATION"); + std::env::remove_var("ABORT_MIGRATION"); } println!("\nMigrator::up"); @@ -151,7 +151,7 @@ async fn run_migration(url: &str, db_name: &str, schema: &str) -> Result<(), DbE println!("\nRoll back changes when encounter errors"); // Set a flag to throw error inside `m20230109_000001_seed_cake_table.rs` - std::env::set_var("ABOARD_MIGRATION", "YES"); + std::env::set_var("ABORT_MIGRATION", "YES"); // Should throw an error println!("\nMigrator::down"); @@ -170,7 +170,7 @@ async fn run_migration(url: &str, db_name: &str, schema: &str) -> Result<(), DbE assert!(manager.has_table("fruit").await?); // Unset the flag - std::env::remove_var("ABOARD_MIGRATION"); + std::env::remove_var("ABORT_MIGRATION"); } println!("\nMigrator::down"); diff --git a/sea-orm-migration/tests/migrator/m20220118_000001_create_cake_table.rs b/sea-orm-migration/tests/migrator/m20220118_000001_create_cake_table.rs index 28bc3e5bf..926ba146d 100644 --- a/sea-orm-migration/tests/migrator/m20220118_000001_create_cake_table.rs +++ b/sea-orm-migration/tests/migrator/m20220118_000001_create_cake_table.rs @@ -28,7 +28,7 @@ impl MigrationTrait for Migration { .drop_table(Table::drop().table(Cake::Table).to_owned()) .await?; - if std::env::var_os("ABOARD_MIGRATION").eq(&Some("YES".into())) { + if std::env::var_os("ABORT_MIGRATION").eq(&Some("YES".into())) { return Err(DbErr::Migration( "Aboard migration and rollback changes".into(), )); diff --git a/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs b/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs index 7f8498252..c902830ba 100644 --- a/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs +++ b/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs @@ -20,7 +20,7 @@ impl MigrationTrait for Migration { transaction.commit().await?; - if std::env::var_os("ABOARD_MIGRATION").eq(&Some("YES".into())) { + if std::env::var_os("ABORT_MIGRATION").eq(&Some("YES".into())) { return Err(DbErr::Migration( "Aboard migration and rollback changes".into(), )); From face72b83f9095929607dd754c9a8db24b2977b3 Mon Sep 17 00:00:00 2001 From: Billy Chan Date: Mon, 30 Jan 2023 12:55:44 +0800 Subject: [PATCH 9/9] abort migration before committing nested transaction --- .../tests/migrator/m20230109_000001_seed_cake_table.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs b/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs index c902830ba..91314f5ef 100644 --- a/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs +++ b/sea-orm-migration/tests/migrator/m20230109_000001_seed_cake_table.rs @@ -18,14 +18,14 @@ impl MigrationTrait for Migration { .insert(&transaction) .await?; - transaction.commit().await?; - if std::env::var_os("ABORT_MIGRATION").eq(&Some("YES".into())) { return Err(DbErr::Migration( "Aboard migration and rollback changes".into(), )); } + transaction.commit().await?; + Ok(()) }