migri/migration.py
async def _apply_migrations(
        self, migrations: List[Migration], dry_run: bool
    ) -> AsyncGenerator[MigrationResult, None]:
        """Apply migrations and yield names of migrations that were applied"""

        # If a migration fails, fail subsequent ones automatically w/out applying them
        migration_failed = False

        async with self._optional_transaction(dry_run):
            for migration in migrations:
                if migration_failed:
                    yield MigrationResult(
                        migration_name=migration.name,
                        message="previous migration failed",
                        status=MigrationStatus.FAILURE,
                    )
                else:
                    async with self._optional_transaction(not dry_run):
                        result = await self._apply(migration)
                        migration_failed = (
                            True if result.status == MigrationStatus.FAILURE else False
                        )

                        if migration_failed:
                            raise self.RollbackTransaction

                    yield result

            raise self.RollbackTransaction

        if dry_run:
            self.echo.info("Successfully applied migrations in dry run mode.")
Apply migrations and yield names of migrations that were applied
Apply migrations and yield names of migrations that were applied
Apply migrations and yield names of migrations that were applied
async def _apply_migrations(
        self, migrations: List[Migration], dry_run: bool
    ) -> AsyncGenerator[MigrationResult, None]:
        migration_failed = False
        async with self._optional_transaction(dry_run):
            for migration in migrations:
                if migration_failed:
                    yield MigrationResult(
                        migration_name=migration.name,
                        message="previous migration failed",
                        status=MigrationStatus.FAILURE,
                    )
                else:
                    async with self._optional_transaction(not dry_run):
                        result = await self._apply(migration)
                        migration_failed = (
                            True if result.status == MigrationStatus.FAILURE else False
                        )
                        if migration_failed:
                            raise self.RollbackTransaction
                    yield result
            raise self.RollbackTransaction
        if dry_run:
            self.echo.info("Successfully applied migrations in dry run mode.")
async def _apply_migrations ( self , migrations : List [ Migration ] , dry_run : bool ) -> AsyncGenerator [ MigrationResult , None ] : migration_failed = False async with self . _optional_transaction ( dry_run ) : for migration in migrations : if migration_failed : yield MigrationResult ( migration_name = migration . name , message = "previous migration failed" , status = MigrationStatus . FAILURE , ) else : async with self . _optional_transaction ( not dry_run ) : result = await self . _apply ( migration ) migration_failed = ( True if result . status == MigrationStatus . FAILURE else False ) if migration_failed : raise self . RollbackTransaction yield result raise self . RollbackTransaction if dry_run : self . echo . info ( "Successfully applied migrations in dry run mode." )
Apply migrations and yield names of migrations that were applied
Apply migrations and yield names of migrations that were applied
"""Apply migrations and yield names of migrations that were applied""" # If a migration fails, fail subsequent ones automatically w/out applying them
self
migrations
List[Migration]
dry_run
bool