Add per-migration transaction control for SeaORM migrations

[tyt2y3]

Add MigrationTrait::use_transaction() to control whether individual migrations
run inside a transaction, replacing the previous all-or-nothing batch transaction
behavior on Postgres.

Changes

Core (sea-orm)

• Add OwnedTransaction(DatabaseTransaction) variant to DatabaseExecutor enum,
  enabling SchemaManager to own a transaction
• Add DatabaseExecutor::is_transaction() method
• Add IntoDatabaseExecutor impl for owned DatabaseTransaction

Migration (sea-orm-migration)

• Add MigrationTrait::use_transaction() -> Option<bool>:
  • None (default): follow backend convention (Postgres = txn, MySQL/SQLite = no txn)
  • Some(true): force transaction on any backend
  • Some(false): disable automatic transaction
• Add SchemaManager::begin() returning an owned transaction
• Refactor exec_up_with / exec_down_with to wrap each migration individually
  based on use_transaction() + backend, instead of batching all migrations

Tests

• Add run_transaction_test covering all three use_transaction modes plus
  failure-with-txn (DDL rolled back) and failure-without-txn (DDL persists)
• Update existing Postgres rollback assertions to match per-migration behavior
• Switch test runtime from async-std to tokio

Example Usage

// Opt out of automatic transaction (e.g. for CREATE INDEX CONCURRENTLY)
impl MigrationTrait for Migration {
    fn use_transaction(&self) -> Option<bool> { Some(false) }

    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
        // DDL in one transaction
        let m = manager.begin().await?;
        m.create_table(...).await?;
        m.commit().await?;

        // Non-transactional DDL
        manager.get_connection()
            .execute_unprepared("CREATE INDEX CONCURRENTLY ...")
            .await?;
        Ok(())
    }
}

[reneklacan]

love it ❤️

[sinder38]

this Is very minor but why not:

async fn insert_migration_record(db: &impl ConnectionTrait, ...)

[tyt2y3]

thanks for the suggestion, a long time ago the convention is to use where T. we can have another PR to migrate all of them in the codebase where it makes sense

[github-actions]

🎉 Released In 2.0.0-rc.36 🎉

Huge thanks for the contribution!
This feature has now been released, so it's a great time to upgrade.
Show some love with a ⭐ on our repo, every star counts!

[yuriy-yarosh]

@tyt2y3 Yeah... but It does wrap on drop even when I've

fn use_transaction(&self) -> Option<bool> {
        Some(false)
    }

Rolling back all applied migrations
Rolling back migration 'm0000006_base_enums'
Migration 'm0000006_base_enums' has been rolled back
Rolling back migration 'm0000005_base_types'
Migration 'm0000005_base_types' has been rolled back
Rolling back migration 'm0000004_extensions'
Migration 'm0000004_extensions' has been rolled back
Rolling back migration 'm0000003_schemas'
Migration 'm0000003_schemas' has been rolled back
Rolling back migration 'm0000002_databases'
Migration 'm0000002_databases' has been rolled back
Rolling back migration 'm0000001_tablespaces'
Execution Error: error returned from database: DROP TABLESPACE неможливо запустити всередині блоку транзакції
Fail to run migration

use sea_orm_migration::{prelude::*, sea_orm::Statement};

#[derive(DeriveMigrationName)]
pub struct DatabaseMigration;

#[async_trait::async_trait]
impl MigrationTrait for DatabaseMigration {
    fn use_transaction(&self) -> Option<bool> {
        Some(false)
    }

    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
        let db = manager.get_connection();

        db.execute_unprepared("CREATE DATABASE app;").await?;

        Ok(())
    }

    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
        let db = manager.get_connection();

        // 1. Terminate all active connections to the target database
        let disconnect_query = format!(
            "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = '{}' AND pid <> pg_backend_pid();",
            "app"
        );
        db.execute_unprepared(&disconnect_query).await?;

        // db.execute_raw(Statement::from_string(db.get_database_backend(), "DROP DATABASE app;")).await?;
        db.execute_unprepared("DROP DATABASE IF EXISTS app;").await?;

        Ok(())
    }
}