diff --git a/sea-orm-migration/src/cli.rs b/sea-orm-migration/src/cli.rs
index a296f949b5..6222009c21 100644
--- a/sea-orm-migration/src/cli.rs
+++ b/sea-orm-migration/src/cli.rs
@@ -8,13 +8,13 @@ use tracing_subscriber::{EnvFilter, prelude::*};
 use sea_orm::{ConnectOptions, Database, DbConn, DbErr};
 use sea_orm_cli::{MigrateSubcommands, run_migrate_generate, run_migrate_init};
 
-use super::MigratorTrait;
+use super::MigratorTraitSelf;
 
 const MIGRATION_DIR: &str = "./";
 
 pub async fn run_cli<M>(migrator: M)
 where
-    M: MigratorTrait,
+    M: MigratorTraitSelf,
 {
     run_cli_with_connection(migrator, Database::connect).await;
 }
@@ -26,7 +26,7 @@ where
 /// extensions.
 pub async fn run_cli_with_connection<M, F, Fut>(migrator: M, make_connection: F)
 where
-    M: MigratorTrait,
+    M: MigratorTraitSelf,
     F: FnOnce(ConnectOptions) -> Fut,
     Fut: Future<Output = Result<DbConn, DbErr>>,
 {
@@ -52,13 +52,13 @@ where
 }
 
 pub async fn run_migrate<M>(
-    _: M,
+    migrator: M,
     db: &DbConn,
     command: Option<MigrateSubcommands>,
     verbose: bool,
 ) -> Result<(), Box<dyn Error>>
 where
-    M: MigratorTrait,
+    M: MigratorTraitSelf,
 {
     let filter = match verbose {
         true => "debug",
@@ -85,19 +85,19 @@ where
     };
 
     match command {
-        Some(MigrateSubcommands::Fresh) => M::fresh(db).await?,
-        Some(MigrateSubcommands::Refresh) => M::refresh(db).await?,
-        Some(MigrateSubcommands::Reset) => M::reset(db).await?,
-        Some(MigrateSubcommands::Status) => M::status(db).await?,
-        Some(MigrateSubcommands::Up { num }) => M::up(db, num).await?,
-        Some(MigrateSubcommands::Down { num }) => M::down(db, Some(num)).await?,
+        Some(MigrateSubcommands::Fresh) => migrator.fresh(db).await?,
+        Some(MigrateSubcommands::Refresh) => migrator.refresh(db).await?,
+        Some(MigrateSubcommands::Reset) => migrator.reset(db).await?,
+        Some(MigrateSubcommands::Status) => migrator.status(db).await?,
+        Some(MigrateSubcommands::Up { num }) => migrator.up(db, num).await?,
+        Some(MigrateSubcommands::Down { num }) => migrator.down(db, Some(num)).await?,
         Some(MigrateSubcommands::Init) => run_migrate_init(MIGRATION_DIR)?,
         Some(MigrateSubcommands::Generate {
             migration_name,
             universal_time: _,
             local_time,
         }) => run_migrate_generate(MIGRATION_DIR, &migration_name, !local_time)?,
-        _ => M::up(db, None).await?,
+        _ => migrator.up(db, None).await?,
     };
 
     Ok(())
diff --git a/sea-orm-migration/src/migrator.rs b/sea-orm-migration/src/migrator.rs
index 01879b55a9..35f03e5be4 100644
--- a/sea-orm-migration/src/migrator.rs
+++ b/sea-orm-migration/src/migrator.rs
@@ -1,24 +1,17 @@
 mod queries;
-use queries::*;
 
-use std::collections::HashSet;
+mod exec;
+use exec::*;
+
+mod with_self;
+pub use with_self::*;
+
 use std::fmt::Display;
-use std::future::Future;
-use std::pin::Pin;
-use std::time::SystemTime;
 use tracing::info;
 
-use sea_orm::sea_query::{
-    Alias, Expr, ExprTrait, ForeignKey, IntoIden, Order, Query, Table, extension::postgres::Type,
-};
-use sea_orm::{
-    ActiveValue, ConnectionTrait, DbBackend, DbErr, DynIden, EntityTrait, FromQueryResult,
-    Iterable, QueryFilter, Schema, Statement, TransactionTrait,
-};
-#[allow(unused_imports)]
-use sea_schema::probe::SchemaProbe;
-
 use super::{IntoSchemaManagerConnection, MigrationTrait, SchemaManager, seaql_migrations};
+use sea_orm::sea_query::IntoIden;
+use sea_orm::{ConnectionTrait, DbBackend, DbErr, DynIden, TransactionTrait};
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
 /// Status of migration
@@ -154,10 +147,7 @@ pub trait MigratorTrait: Send {
     where
         C: IntoSchemaManagerConnection<'c>,
     {
-        exec_with_connection::<'_, _, _>(db, move |manager| {
-            Box::pin(async move { exec_fresh::<Self>(manager).await })
-        })
-        .await
+        exec_with_connection!(db, async |manager| { exec_fresh::<Self>(manager).await }).await
     }
 
     /// Rollback all applied migrations, then reapply all migrations
@@ -165,11 +155,9 @@ pub trait MigratorTrait: Send {
     where
         C: IntoSchemaManagerConnection<'c>,
     {
-        exec_with_connection::<'_, _, _>(db, move |manager| {
-            Box::pin(async move {
-                exec_down::<Self>(manager, None).await?;
-                exec_up::<Self>(manager, None).await
-            })
+        exec_with_connection!(db, async |manager| {
+            exec_down::<Self>(manager, None).await?;
+            exec_up::<Self>(manager, None).await
         })
         .await
     }
@@ -179,16 +167,12 @@ pub trait MigratorTrait: Send {
     where
         C: IntoSchemaManagerConnection<'c>,
     {
-        exec_with_connection::<'_, _, _>(db, move |manager| {
-            Box::pin(async move {
-                // Rollback all applied migrations first
-                exec_down::<Self>(manager, None).await?;
-
-                // Then drop the migration table itself
-                uninstall(manager, Self::migration_table_name()).await?;
+        exec_with_connection!(db, async |manager| {
+            // Rollback all applied migrations first
+            exec_down::<Self>(manager, None).await?;
 
-                Ok(())
-            })
+            // Then drop the migration table itself
+            uninstall(manager, Self::migration_table_name()).await
         })
         .await
     }
@@ -199,8 +183,8 @@ pub trait MigratorTrait: Send {
     where
         C: IntoSchemaManagerConnection<'c>,
     {
-        exec_with_connection::<'_, _, _>(db, move |manager| {
-            Box::pin(uninstall(manager, Self::migration_table_name()))
+        exec_with_connection!(db, async |manager| {
+            uninstall(manager, Self::migration_table_name()).await
         })
         .await
     }
@@ -210,8 +194,8 @@ pub trait MigratorTrait: Send {
     where
         C: IntoSchemaManagerConnection<'c>,
     {
-        exec_with_connection::<'_, _, _>(db, move |manager| {
-            Box::pin(async move { exec_up::<Self>(manager, steps).await })
+        exec_with_connection!(db, async |manager| {
+            exec_up::<Self>(manager, steps).await
         })
         .await
     }
@@ -221,118 +205,13 @@ pub trait MigratorTrait: Send {
     where
         C: IntoSchemaManagerConnection<'c>,
     {
-        exec_with_connection::<'_, _, _>(db, move |manager| {
-            Box::pin(async move { exec_down::<Self>(manager, steps).await })
+        exec_with_connection!(db, async |manager| {
+            exec_down::<Self>(manager, steps).await
         })
         .await
     }
 }
 
-async fn get_migration_models<C>(
-    db: &C,
-    migration_table_name: DynIden,
-) -> Result<Vec<seaql_migrations::Model>, DbErr>
-where
-    C: ConnectionTrait,
-{
-    let stmt = Query::select()
-        .table_name(migration_table_name)
-        .columns(seaql_migrations::Column::iter().map(IntoIden::into_iden))
-        .order_by(seaql_migrations::Column::Version, Order::Asc)
-        .to_owned();
-    let builder = db.get_database_backend();
-    seaql_migrations::Model::find_by_statement(builder.build(&stmt))
-        .all(db)
-        .await
-}
-
-fn get_migration_with_status(
-    migration_files: Vec<Migration>,
-    migration_models: Vec<seaql_migrations::Model>,
-) -> Result<Vec<Migration>, DbErr> {
-    let mut migration_files = migration_files;
-
-    let migration_in_db: HashSet<String> = migration_models
-        .into_iter()
-        .map(|model| model.version)
-        .collect();
-    let migration_in_fs: HashSet<String> = migration_files
-        .iter()
-        .map(|file| file.migration.name().to_string())
-        .collect();
-
-    let pending_migrations = &migration_in_fs - &migration_in_db;
-    for migration_file in migration_files.iter_mut() {
-        if !pending_migrations.contains(migration_file.migration.name()) {
-            migration_file.status = MigrationStatus::Applied;
-        }
-    }
-
-    let missing_migrations_in_fs = &migration_in_db - &migration_in_fs;
-    let errors: Vec<String> = missing_migrations_in_fs
-            .iter()
-            .map(|missing_migration| {
-                format!("Migration file of version '{missing_migration}' is missing, this migration has been applied but its file is missing")
-            }).collect();
-
-    if !errors.is_empty() {
-        Err(DbErr::Custom(errors.join("\n")))
-    } else {
-        Ok(migration_files)
-    }
-}
-
-async fn exec_with_connection<'c, C, F>(db: C, f: F) -> Result<(), DbErr>
-where
-    C: IntoSchemaManagerConnection<'c>,
-    F: for<'b> Fn(
-        &'b SchemaManager<'_>,
-    ) -> Pin<Box<dyn Future<Output = Result<(), DbErr>> + Send + 'b>>,
-{
-    let db = db.into_database_executor();
-
-    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
-        }
-        db => Err(DbErr::BackendNotSupported {
-            db: db.as_str(),
-            ctx: "exec_with_connection",
-        }),
-    }
-}
-
-async fn install<C>(db: &C, migration_table_name: DynIden) -> 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)
-        .table_name(migration_table_name);
-    stmt.if_not_exists();
-    db.execute(&stmt).await?;
-    Ok(())
-}
-
-async fn uninstall(
-    manager: &SchemaManager<'_>,
-    migration_table_name: DynIden,
-) -> Result<(), DbErr> {
-    let mut stmt = Table::drop();
-    stmt.table(migration_table_name).if_exists().cascade();
-    manager.drop_table(stmt).await?;
-    Ok(())
-}
-
 async fn exec_fresh<M>(manager: &SchemaManager<'_>) -> Result<(), DbErr>
 where
     M: MigratorTrait + ?Sized,
@@ -346,84 +225,6 @@ where
     exec_up::<M>(manager, None).await
 }
 
-async fn drop_everything<C: ConnectionTrait>(db: &C) -> Result<(), DbErr> {
-    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_raw(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(&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(&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(&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(&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(&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));
-            db.execute(&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_raw(Statement::from_string(
-            db_backend,
-            "PRAGMA foreign_keys = ON".to_owned(),
-        ))
-        .await?;
-        info!("Foreign key check restored");
-    }
-
-    Ok(())
-}
-
 async fn exec_up<M>(manager: &SchemaManager<'_>, steps: Option<u32>) -> Result<(), DbErr>
 where
     M: MigratorTrait + ?Sized,
@@ -441,48 +242,6 @@ where
     .await
 }
 
-async fn exec_up_with(
-    manager: &SchemaManager<'_>,
-    mut steps: Option<u32>,
-    pending_migrations: Vec<Migration>,
-    migration_table_name: DynIden,
-) -> Result<(), DbErr> {
-    let db = manager.get_connection();
-
-    if let Some(steps) = steps {
-        info!("Applying {} pending migrations", steps);
-    } else {
-        info!("Applying all pending migrations");
-    }
-    if pending_migrations.is_empty() {
-        info!("No pending migrations");
-    }
-
-    for Migration { migration, .. } in pending_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::Entity::insert(seaql_migrations::ActiveModel {
-            version: ActiveValue::Set(migration.name().to_owned()),
-            applied_at: ActiveValue::Set(now.as_secs() as i64),
-        })
-        .table_name(migration_table_name.clone())
-        .exec(db)
-        .await?;
-    }
-
-    Ok(())
-}
-
 async fn exec_down<M>(manager: &SchemaManager<'_>, steps: Option<u32>) -> Result<(), DbErr>
 where
     M: MigratorTrait + ?Sized,
@@ -499,40 +258,3 @@ where
     )
     .await
 }
-
-async fn exec_down_with(
-    manager: &SchemaManager<'_>,
-    mut steps: Option<u32>,
-    applied_migrations: Vec<Migration>,
-    migration_table_name: DynIden,
-) -> Result<(), DbErr> {
-    let db = manager.get_connection();
-
-    if let Some(steps) = steps {
-        info!("Rolling back {} applied migrations", steps);
-    } else {
-        info!("Rolling back all applied migrations");
-    }
-    if applied_migrations.is_empty() {
-        info!("No applied migrations");
-    }
-
-    for Migration { migration, .. } in applied_migrations.into_iter().rev() {
-        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(Expr::col(seaql_migrations::Column::Version).eq(migration.name()))
-            .table_name(migration_table_name.clone())
-            .exec(db)
-            .await?;
-    }
-
-    Ok(())
-}
diff --git a/sea-orm-migration/src/migrator/exec.rs b/sea-orm-migration/src/migrator/exec.rs
new file mode 100644
index 0000000000..314eb9a279
--- /dev/null
+++ b/sea-orm-migration/src/migrator/exec.rs
@@ -0,0 +1,275 @@
+use std::collections::HashSet;
+use std::time::SystemTime;
+use tracing::info;
+
+use super::{Migration, MigrationStatus, queries::*};
+use crate::{SchemaManager, seaql_migrations};
+use sea_orm::sea_query::{
+    Alias, Expr, ExprTrait, ForeignKey, IntoIden, Order, Query, Table, extension::postgres::Type,
+};
+use sea_orm::{
+    ActiveValue, ConnectionTrait, DbBackend, DbErr, DynIden, EntityTrait, FromQueryResult,
+    Iterable, QueryFilter, Schema, Statement,
+};
+
+pub async fn get_migration_models<C>(
+    db: &C,
+    migration_table_name: DynIden,
+) -> Result<Vec<seaql_migrations::Model>, DbErr>
+where
+    C: ConnectionTrait,
+{
+    let stmt = Query::select()
+        .table_name(migration_table_name)
+        .columns(seaql_migrations::Column::iter().map(IntoIden::into_iden))
+        .order_by(seaql_migrations::Column::Version, Order::Asc)
+        .to_owned();
+    let builder = db.get_database_backend();
+    seaql_migrations::Model::find_by_statement(builder.build(&stmt))
+        .all(db)
+        .await
+}
+
+pub fn get_migration_with_status(
+    migration_files: Vec<Migration>,
+    migration_models: Vec<seaql_migrations::Model>,
+) -> Result<Vec<Migration>, DbErr> {
+    let mut migration_files = migration_files;
+
+    let migration_in_db: HashSet<String> = migration_models
+        .into_iter()
+        .map(|model| model.version)
+        .collect();
+    let migration_in_fs: HashSet<String> = migration_files
+        .iter()
+        .map(|file| file.migration.name().to_string())
+        .collect();
+
+    let pending_migrations = &migration_in_fs - &migration_in_db;
+    for migration_file in migration_files.iter_mut() {
+        if !pending_migrations.contains(migration_file.migration.name()) {
+            migration_file.status = MigrationStatus::Applied;
+        }
+    }
+
+    let missing_migrations_in_fs = &migration_in_db - &migration_in_fs;
+    let errors: Vec<String> = missing_migrations_in_fs
+            .iter()
+            .map(|missing_migration| {
+                format!("Migration file of version '{missing_migration}' is missing, this migration has been applied but its file is missing")
+            }).collect();
+
+    if !errors.is_empty() {
+        Err(DbErr::Custom(errors.join("\n")))
+    } else {
+        Ok(migration_files)
+    }
+}
+
+macro_rules! exec_with_connection {
+    ($db:ident, $fn:expr) => {{
+        async {
+            let db = $db.into_database_executor();
+
+            match db.get_database_backend() {
+                DbBackend::Postgres => {
+                    let transaction = db.begin().await?;
+                    let manager = SchemaManager::new(&transaction);
+                    $fn(&manager).await?;
+                    transaction.commit().await
+                }
+                DbBackend::MySql | DbBackend::Sqlite => {
+                    let manager = SchemaManager::new(db);
+                    $fn(&manager).await
+                }
+                db => Err(DbErr::BackendNotSupported {
+                    db: db.as_str(),
+                    ctx: "exec_with_connection",
+                }),
+            }
+        }
+    }};
+}
+
+pub(crate) use exec_with_connection;
+
+pub async fn install<C>(db: &C, migration_table_name: DynIden) -> 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)
+        .table_name(migration_table_name);
+    stmt.if_not_exists();
+    db.execute(&stmt).await?;
+    Ok(())
+}
+
+pub async fn uninstall(
+    manager: &SchemaManager<'_>,
+    migration_table_name: DynIden,
+) -> Result<(), DbErr> {
+    let mut stmt = Table::drop();
+    stmt.table(migration_table_name).if_exists().cascade();
+    manager.drop_table(stmt).await?;
+    Ok(())
+}
+
+pub async fn drop_everything<C: ConnectionTrait>(db: &C) -> Result<(), DbErr> {
+    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_raw(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(&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(&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(&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(&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(&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));
+            db.execute(&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_raw(Statement::from_string(
+            db_backend,
+            "PRAGMA foreign_keys = ON".to_owned(),
+        ))
+        .await?;
+        info!("Foreign key check restored");
+    }
+
+    Ok(())
+}
+
+pub async fn exec_up_with(
+    manager: &SchemaManager<'_>,
+    mut steps: Option<u32>,
+    pending_migrations: Vec<Migration>,
+    migration_table_name: DynIden,
+) -> Result<(), DbErr> {
+    let db = manager.get_connection();
+
+    if let Some(steps) = steps {
+        info!("Applying {} pending migrations", steps);
+    } else {
+        info!("Applying all pending migrations");
+    }
+    if pending_migrations.is_empty() {
+        info!("No pending migrations");
+    }
+
+    for Migration { migration, .. } in pending_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::Entity::insert(seaql_migrations::ActiveModel {
+            version: ActiveValue::Set(migration.name().to_owned()),
+            applied_at: ActiveValue::Set(now.as_secs() as i64),
+        })
+        .table_name(migration_table_name.clone())
+        .exec(db)
+        .await?;
+    }
+
+    Ok(())
+}
+
+pub async fn exec_down_with(
+    manager: &SchemaManager<'_>,
+    mut steps: Option<u32>,
+    applied_migrations: Vec<Migration>,
+    migration_table_name: DynIden,
+) -> Result<(), DbErr> {
+    let db = manager.get_connection();
+
+    if let Some(steps) = steps {
+        info!("Rolling back {} applied migrations", steps);
+    } else {
+        info!("Rolling back all applied migrations");
+    }
+    if applied_migrations.is_empty() {
+        info!("No applied migrations");
+    }
+
+    for Migration { migration, .. } in applied_migrations.into_iter().rev() {
+        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(Expr::col(seaql_migrations::Column::Version).eq(migration.name()))
+            .table_name(migration_table_name.clone())
+            .exec(db)
+            .await?;
+    }
+
+    Ok(())
+}
diff --git a/sea-orm-migration/src/migrator/with_self.rs b/sea-orm-migration/src/migrator/with_self.rs
new file mode 100644
index 0000000000..2949574702
--- /dev/null
+++ b/sea-orm-migration/src/migrator/with_self.rs
@@ -0,0 +1,326 @@
+use super::{Migration, MigrationStatus, exec::*};
+use crate::{IntoSchemaManagerConnection, MigrationTrait, SchemaManager, seaql_migrations};
+use sea_orm::sea_query::IntoIden;
+use sea_orm::{ConnectionTrait, DbBackend, DbErr, DynIden, TransactionTrait};
+
+use tracing::info;
+
+/// Performing migrations on a database
+#[async_trait::async_trait]
+pub trait MigratorTraitSelf: Sized + Send + Sync {
+    /// Vector of migrations in time sequence
+    fn migrations(&self) -> Vec<Box<dyn MigrationTrait>>;
+
+    /// Name of the migration table, it is `seaql_migrations` by default
+    fn migration_table_name(&self) -> DynIden {
+        seaql_migrations::Entity.into_iden()
+    }
+
+    /// Get list of migrations wrapped in `Migration` struct
+    fn get_migration_files(&self) -> Vec<Migration> {
+        self.migrations()
+            .into_iter()
+            .map(|migration| Migration {
+                migration,
+                status: MigrationStatus::Pending,
+            })
+            .collect()
+    }
+
+    /// Get list of applied migrations from database
+    async fn get_migration_models<C>(&self, db: &C) -> Result<Vec<seaql_migrations::Model>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        self.install(db).await?;
+        get_migration_models(db, self.migration_table_name()).await
+    }
+
+    /// Get list of migrations with status
+    async fn get_migration_with_status<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        self.install(db).await?;
+        get_migration_with_status(
+            self.get_migration_files(),
+            self.get_migration_models(db).await?,
+        )
+    }
+
+    /// Get list of pending migrations
+    async fn get_pending_migrations<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        self.install(db).await?;
+        Ok(self
+            .get_migration_with_status(db)
+            .await?
+            .into_iter()
+            .filter(|file| file.status == MigrationStatus::Pending)
+            .collect())
+    }
+
+    /// Get list of applied migrations
+    async fn get_applied_migrations<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        self.install(db).await?;
+        Ok(self
+            .get_migration_with_status(db)
+            .await?
+            .into_iter()
+            .filter(|file| file.status == MigrationStatus::Applied)
+            .collect())
+    }
+
+    /// Create migration table `seaql_migrations` in the database
+    async fn install<C>(&self, db: &C) -> Result<(), DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        install(db, self.migration_table_name()).await
+    }
+
+    /// Check the status of all migrations
+    async fn status<C>(&self, db: &C) -> Result<(), DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        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(())
+    }
+
+    /// Drop all tables from the database, then reapply all migrations
+    async fn fresh<'c, C>(&self, db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection!(db, async |manager| { exec_fresh(self, manager).await }).await
+    }
+
+    /// Rollback all applied migrations, then reapply all migrations
+    async fn refresh<'c, C>(&self, db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection!(db, async |manager| {
+            exec_down(self, manager, None).await?;
+            exec_up(self, manager, None).await
+        })
+        .await
+    }
+
+    /// Rollback all applied migrations
+    async fn reset<'c, C>(&self, db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection!(db, async |manager| {
+            // Rollback all applied migrations first
+            exec_down(self, manager, None).await?;
+
+            // Then drop the migration table itself
+            uninstall(manager, self.migration_table_name()).await
+        })
+        .await
+    }
+
+    /// Uninstall migration tracking table only (non-destructive)
+    /// This will drop the `seaql_migrations` table but won't rollback other schema changes.
+    async fn uninstall<'c, C>(&self, db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection!(db, async |manager| {
+            uninstall(manager, self.migration_table_name()).await
+        })
+        .await
+    }
+
+    /// Apply pending migrations
+    async fn up<'c, C>(&self, db: C, steps: Option<u32>) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection!(db, async |manager| { exec_up(self, manager, steps).await }).await
+    }
+
+    /// Rollback applied migrations
+    async fn down<'c, C>(&self, db: C, steps: Option<u32>) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection!(db, async |manager| {
+            exec_down(self, manager, steps).await
+        })
+        .await
+    }
+}
+
+#[async_trait::async_trait]
+impl<M> MigratorTraitSelf for M
+where
+    M: super::MigratorTrait + Sized + Send + Sync,
+{
+    fn migrations(&self) -> Vec<Box<dyn MigrationTrait>> {
+        M::migrations()
+    }
+
+    fn migration_table_name(&self) -> DynIden {
+        M::migration_table_name()
+    }
+
+    fn get_migration_files(&self) -> Vec<Migration> {
+        M::get_migration_files()
+    }
+
+    async fn get_migration_models<C>(&self, db: &C) -> Result<Vec<seaql_migrations::Model>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        M::get_migration_models(db).await
+    }
+
+    async fn get_migration_with_status<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        M::get_migration_with_status(db).await
+    }
+
+    async fn get_pending_migrations<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        M::get_pending_migrations(db).await
+    }
+
+    async fn get_applied_migrations<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        M::get_applied_migrations(db).await
+    }
+
+    async fn install<C>(&self, db: &C) -> Result<(), DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        M::install(db).await
+    }
+
+    /// Check the status of all migrations
+    async fn status<C>(&self, db: &C) -> Result<(), DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        M::status(db).await
+    }
+
+    async fn fresh<'c, C>(&self, db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        M::fresh(db).await
+    }
+
+    async fn refresh<'c, C>(&self, db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        M::refresh(db).await
+    }
+
+    async fn reset<'c, C>(&self, db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        M::reset(db).await
+    }
+
+    async fn uninstall<'c, C>(&self, db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        M::uninstall(db).await
+    }
+
+    async fn up<'c, C>(&self, db: C, steps: Option<u32>) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        M::up(db, steps).await
+    }
+
+    async fn down<'c, C>(&self, db: C, steps: Option<u32>) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        M::down(db, steps).await
+    }
+}
+
+async fn exec_fresh<M>(migrator: &M, manager: &SchemaManager<'_>) -> Result<(), DbErr>
+where
+    M: MigratorTraitSelf,
+{
+    let db = manager.get_connection();
+
+    migrator.install(db).await?;
+
+    drop_everything(db).await?;
+
+    exec_up(migrator, manager, None).await
+}
+
+async fn exec_up<M>(
+    migrator: &M,
+    manager: &SchemaManager<'_>,
+    steps: Option<u32>,
+) -> Result<(), DbErr>
+where
+    M: MigratorTraitSelf,
+{
+    let db = manager.get_connection();
+
+    migrator.install(db).await?;
+
+    exec_up_with(
+        manager,
+        steps,
+        migrator.get_pending_migrations(db).await?,
+        migrator.migration_table_name(),
+    )
+    .await
+}
+
+async fn exec_down<M>(
+    migrator: &M,
+    manager: &SchemaManager<'_>,
+    steps: Option<u32>,
+) -> Result<(), DbErr>
+where
+    M: MigratorTraitSelf,
+{
+    let db = manager.get_connection();
+
+    migrator.install(db).await?;
+
+    exec_down_with(
+        manager,
+        steps,
+        migrator.get_applied_migrations(db).await?,
+        migrator.migration_table_name(),
+    )
+    .await
+}
diff --git a/sea-orm-migration/tests/common/migrator/mod.rs b/sea-orm-migration/tests/common/migrator/mod.rs
index 772da4927d..d2e3727438 100644
--- a/sea-orm-migration/tests/common/migrator/mod.rs
+++ b/sea-orm-migration/tests/common/migrator/mod.rs
@@ -1,2 +1,3 @@
 pub mod default;
 pub mod override_migration_table_name;
+pub mod with_self;
diff --git a/sea-orm-migration/tests/common/migrator/with_self.rs b/sea-orm-migration/tests/common/migrator/with_self.rs
new file mode 100644
index 0000000000..5734817fe2
--- /dev/null
+++ b/sea-orm-migration/tests/common/migrator/with_self.rs
@@ -0,0 +1,20 @@
+use crate::common::migration::*;
+use sea_orm_migration::{MigratorTraitSelf, prelude::*};
+
+pub struct Migrator {
+    pub i: i32,
+}
+
+#[async_trait::async_trait]
+impl MigratorTraitSelf for Migrator {
+    fn migrations(&self) -> Vec<Box<dyn MigrationTrait>> {
+        vec![
+            Box::new(m20220118_000001_create_cake_table::Migration),
+            Box::new(m20220118_000002_create_fruit_table::Migration),
+            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),
+        ]
+    }
+}
diff --git a/sea-orm-migration/tests/main.rs b/sea-orm-migration/tests/main.rs
index fbb10a0cb9..25537fdff7 100644
--- a/sea-orm-migration/tests/main.rs
+++ b/sea-orm-migration/tests/main.rs
@@ -2,7 +2,7 @@ mod common;
 
 use common::migrator::*;
 use sea_orm::{ConnectOptions, ConnectionTrait, Database, DbBackend, DbErr, Statement};
-use sea_orm_migration::{migrator::MigrationStatus, prelude::*};
+use sea_orm_migration::{MigratorTraitSelf, migrator::MigrationStatus, prelude::*};
 
 #[async_std::test]
 async fn main() -> Result<(), DbErr> {
@@ -22,6 +22,14 @@ async fn main() -> Result<(), DbErr> {
     )
     .await?;
 
+    run_migration(
+        url,
+        with_self::Migrator { i: 12 },
+        "sea_orm_migration",
+        "public",
+    )
+    .await?;
+
     run_migration(
         url,
         override_migration_table_name::Migrator,
@@ -40,14 +48,9 @@ async fn main() -> Result<(), DbErr> {
     Ok(())
 }
 
-async fn run_migration<Migrator>(
-    url: &str,
-    _: Migrator,
-    db_name: &str,
-    schema: &str,
-) -> Result<(), DbErr>
+async fn run_migration<M>(url: &str, migrator: M, db_name: &str, schema: &str) -> Result<(), DbErr>
 where
-    Migrator: MigratorTrait,
+    M: MigratorTraitSelf,
 {
     let db_connect = |url: String| async {
         let connect_options = ConnectOptions::new(url)
@@ -104,12 +107,12 @@ where
     let manager = SchemaManager::new(db);
 
     println!("\nMigrator::status");
-    Migrator::status(db).await?;
+    migrator.status(db).await?;
 
     println!("\nMigrator::install");
-    Migrator::install(db).await?;
+    migrator.install(db).await?;
 
-    let migration_table_name = Migrator::migration_table_name().to_string();
+    let migration_table_name = migrator.migration_table_name().to_string();
     let migration_table_name = migration_table_name.as_str();
     assert!(manager.has_table(migration_table_name).await?);
     if migration_table_name != "seaql_migrations" {
@@ -117,22 +120,22 @@ where
     }
 
     println!("\nMigrator::reset");
-    Migrator::reset(db).await?;
+    migrator.reset(db).await?;
 
     assert!(!manager.has_table("cake").await?);
     assert!(!manager.has_table("fruit").await?);
 
     println!("\nMigrator::up");
-    Migrator::up(db, Some(0)).await?;
+    migrator.up(db, Some(0)).await?;
 
     assert!(!manager.has_table("cake").await?);
     assert!(!manager.has_table("fruit").await?);
 
     println!("\nMigrator::up");
-    Migrator::up(db, Some(1)).await?;
+    migrator.up(db, Some(1)).await?;
 
     println!("\nMigrator::get_pending_migrations");
-    let migrations = Migrator::get_pending_migrations(db).await?;
+    let migrations = migrator.get_pending_migrations(db).await?;
     assert_eq!(migrations.len(), 5);
 
     let migration = migrations.get(0).unwrap();
@@ -143,13 +146,13 @@ where
     assert!(!manager.has_table("fruit").await?);
 
     println!("\nMigrator::down");
-    Migrator::down(db, Some(0)).await?;
+    migrator.down(db, Some(0)).await?;
 
     assert!(manager.has_table("cake").await?);
     assert!(!manager.has_table("fruit").await?);
 
     println!("\nMigrator::down");
-    Migrator::down(db, Some(1)).await?;
+    migrator.down(db, Some(1)).await?;
 
     assert!(!manager.has_table("cake").await?);
     assert!(!manager.has_table("fruit").await?);
@@ -166,14 +169,14 @@ where
         // Should throw an error
         println!("\nMigrator::up");
         assert_eq!(
-            Migrator::up(db, None).await,
+            migrator.up(db, None).await,
             Err(DbErr::Migration(
                 "Abort migration and rollback changes".into()
             ))
         );
 
         println!("\nMigrator::status");
-        Migrator::status(db).await?;
+        migrator.status(db).await?;
 
         // Check migrations have been rolled back
         assert!(!manager.has_table("cake").await?);
@@ -186,10 +189,10 @@ where
     }
 
     println!("\nMigrator::up");
-    Migrator::up(db, None).await?;
+    migrator.up(db, None).await?;
 
     println!("\nMigrator::get_applied_migrations");
-    let migrations = Migrator::get_applied_migrations(db).await?;
+    let migrations = migrator.get_applied_migrations(db).await?;
     assert_eq!(migrations.len(), 6);
 
     assert!(!manager.has_index("cake", "non_existent_index").await?);
@@ -200,7 +203,7 @@ where
     assert_eq!(migration.status(), MigrationStatus::Applied);
 
     println!("\nMigrator::status");
-    Migrator::status(db).await?;
+    migrator.status(db).await?;
 
     assert!(manager.has_table("cake").await?);
     assert!(manager.has_table("fruit").await?);
@@ -220,14 +223,14 @@ where
         // Should throw an error
         println!("\nMigrator::down");
         assert_eq!(
-            Migrator::down(db, None).await,
+            migrator.down(db, None).await,
             Err(DbErr::Migration(
                 "Abort migration and rollback changes".into()
             ))
         );
 
         println!("\nMigrator::status");
-        Migrator::status(db).await?;
+        migrator.status(db).await?;
 
         // Check migrations have been rolled back
         assert!(manager.has_table("cake").await?);
@@ -240,7 +243,7 @@ where
     }
 
     println!("\nMigrator::down");
-    Migrator::down(db, None).await?;
+    migrator.down(db, None).await?;
 
     assert!(manager.has_table(migration_table_name).await?);
     if migration_table_name != "seaql_migrations" {
@@ -251,25 +254,25 @@ where
     assert!(!manager.has_table("fruit").await?);
 
     println!("\nMigrator::fresh");
-    Migrator::fresh(db).await?;
+    migrator.fresh(db).await?;
 
     assert!(manager.has_table("cake").await?);
     assert!(manager.has_table("fruit").await?);
 
     println!("\nMigrator::refresh");
-    Migrator::refresh(db).await?;
+    migrator.refresh(db).await?;
 
     assert!(manager.has_table("cake").await?);
     assert!(manager.has_table("fruit").await?);
 
     println!("\nMigrator::reset");
-    Migrator::reset(db).await?;
+    migrator.reset(db).await?;
 
     assert!(!manager.has_table("cake").await?);
     assert!(!manager.has_table("fruit").await?);
 
     println!("\nMigrator::status");
-    Migrator::status(db).await?;
+    migrator.status(db).await?;
 
     Ok(())
 }