sea_orm_migration/migrator/

with_self.rs

use super::{Migration, MigrationStatus, exec::*};
use crate::{IntoSchemaManagerConnection, MigrationTrait, SchemaManager, seaql_migrations};
use sea_orm::sea_query::IntoIden;
use sea_orm::{ConnectionTrait, DbErr, DynIden};

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())
    }

    /// Get list of migrations with status, without creating the migration table.
    ///
    /// Unlike [`get_migration_with_status`](Self::get_migration_with_status), this never runs
    /// `CREATE TABLE`, so it can be called by a database user without DDL privileges (for
    /// example, checking pending migrations from a read-only connection). If the migration
    /// table does not exist, every migration is reported as [`MigrationStatus::Pending`].
    async fn get_migration_with_status_read_only<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
    where
        C: ConnectionTrait,
    {
        get_migration_with_status(
            self.get_migration_files(),
            get_migration_models_read_only(db, self.migration_table_name()).await?,
        )
    }

    /// Get list of pending migrations without creating the migration table.
    ///
    /// The read-only counterpart of [`get_pending_migrations`](Self::get_pending_migrations);
    /// see [`get_migration_with_status_read_only`](Self::get_migration_with_status_read_only).
    async fn get_pending_migrations_read_only<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
    where
        C: ConnectionTrait,
    {
        Ok(self
            .get_migration_with_status_read_only(db)
            .await?
            .into_iter()
            .filter(|file| file.status == MigrationStatus::Pending)
            .collect())
    }

    /// Get list of applied migrations without creating the migration table.
    ///
    /// The read-only counterpart of [`get_applied_migrations`](Self::get_applied_migrations);
    /// see [`get_migration_with_status_read_only`](Self::get_migration_with_status_read_only).
    async fn get_applied_migrations_read_only<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
    where
        C: ConnectionTrait,
    {
        Ok(self
            .get_migration_with_status_read_only(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>,
    {
        let db = db.into_database_executor();
        let manager = SchemaManager::new(db);
        exec_fresh(self, &manager).await
    }

    /// Rollback all applied migrations, then reapply all migrations
    async fn refresh<'c, C>(&self, db: C) -> Result<(), DbErr>
    where
        C: IntoSchemaManagerConnection<'c>,
    {
        let db = db.into_database_executor();
        let manager = SchemaManager::new(db);
        exec_down(self, &manager, None).await?;
        exec_up(self, &manager, None).await
    }

    /// Rollback all applied migrations
    async fn reset<'c, C>(&self, db: C) -> Result<(), DbErr>
    where
        C: IntoSchemaManagerConnection<'c>,
    {
        let db = db.into_database_executor();
        let manager = SchemaManager::new(db);
        exec_down(self, &manager, None).await?;
        uninstall(&manager, self.migration_table_name()).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>,
    {
        let db = db.into_database_executor();
        let manager = SchemaManager::new(db);
        uninstall(&manager, self.migration_table_name()).await
    }

    /// Apply pending migrations
    async fn up<'c, C>(&self, db: C, steps: Option<u32>) -> Result<(), DbErr>
    where
        C: IntoSchemaManagerConnection<'c>,
    {
        let db = db.into_database_executor();
        let manager = SchemaManager::new(db);
        exec_up(self, &manager, steps).await
    }

    /// Rollback applied migrations
    async fn down<'c, C>(&self, db: C, steps: Option<u32>) -> Result<(), DbErr>
    where
        C: IntoSchemaManagerConnection<'c>,
    {
        let db = db.into_database_executor();
        let manager = SchemaManager::new(db);
        exec_down(self, &manager, steps).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_migration_with_status_read_only<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
    where
        C: ConnectionTrait,
    {
        M::get_migration_with_status_read_only(db).await
    }

    async fn get_pending_migrations_read_only<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
    where
        C: ConnectionTrait,
    {
        M::get_pending_migrations_read_only(db).await
    }

    async fn get_applied_migrations_read_only<C>(&self, db: &C) -> Result<Vec<Migration>, DbErr>
    where
        C: ConnectionTrait,
    {
        M::get_applied_migrations_read_only(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
}