Alembic

autogenerate(context, message='')

Autogenerate new version with message title.

Similar to the manage.py makemigration command in Django. But you always have to check generated versions before upgrade and fix it manually if it's necessary.

Source code in saritasa_invocations/alembic.py

@invoke.task
def autogenerate(
    context: invoke.Context,
    message: str = "",
) -> None:
    """Autogenerate new version with `message` title.

    Similar to the `manage.py makemigration` command in Django. But you always
    have to check generated versions before upgrade and fix it manually
    if it's necessary.

    """
    if not message:
        raise invoke.Exit(
            code=1,
            message=(
                'Please, use `-m "Version message"` to set message '
                "for autogenerated version."
            ),
        )
    printing.print_success("Autogenerate migrations")
    config = _config.Config.from_context(context)
    rev_id = str(
        len(_get_migration_files_paths(config.alembic.migrations_folder)) + 1,
    )
    rev_id = rev_id.rjust(4, "0")
    command = (
        f'revision --autogenerate --message "{message}" --rev-id={rev_id}'
    )
    run(context, command=command)
    printing.print_success(
        "Migrations generated, please verify them and remove alembic messages",
    )

backup_local_db(context, file='', env_file_path='.env')

Back up local db.

Source code in saritasa_invocations/alembic.py

@invoke.task
def backup_local_db(
    context: invoke.Context,
    file: str = "",
    env_file_path: str = ".env",
) -> None:
    """Back up local db."""
    db.backup_local_db(
        context,
        file=file,
        **_load_local_env_db_settings(context, file=env_file_path),
    )

backup_remote_db(context, file='')

Make dump of remote db and download it.

Source code in saritasa_invocations/alembic.py

@invoke.task
def backup_remote_db(
    context: invoke.Context,
    file: str = "",
) -> str:
    """Make dump of remote db and download it."""
    settings = _load_remote_env_db_settings(context)
    db_k8s.create_dump(
        context,
        file=file,
        **settings,
    )
    return db_k8s.get_dump(
        context,
        file=file,
    )

check_for_adjust_messages(context)

Check migration files for adjust messages.

Source code in saritasa_invocations/alembic.py

@invoke.task
def check_for_adjust_messages(
    context: invoke.Context,
) -> None:
    """Check migration files for adjust messages."""
    printing.print_success("Checking migration files for adjust messages")
    config = _config.Config.from_context(context)
    files_to_clean = []
    for filepath in _get_migration_files_paths(
        config.alembic.migrations_folder,
    ):
        with pathlib.Path(filepath).open() as migration_file:
            file_text = migration_file.read()
            for adjust_message in config.alembic.adjust_messages:
                if adjust_message in file_text:
                    files_to_clean.append(str(filepath))
                    break

    if files_to_clean:
        log_files_msg = "\n\t".join(files_to_clean)
        log_messages = "\n".join(config.alembic.adjust_messages)
        printing.print_error(
            f"Adjust messages found in these migration files:\n"
            f"\t{log_files_msg}\n"
            "Ensure that these files do not contain the following:\n"
            f"{log_messages}",
        )
        raise invoke.Exit(code=1)

check_for_migrations(context)

Check if new migration can be generated.

Source code in saritasa_invocations/alembic.py

@invoke.task
def check_for_migrations(
    context: invoke.Context,
) -> None:
    """Check if new migration can be generated."""
    printing.print_success("Checking for migration")
    run(context, command="check")

downgrade(context, version='base')

Downgrade database.

Use -v version to downgrade to the passed version.

Similar to the manage.py migrate command in Django. By default downgrade to base (similar to migrate zero).

Source code in saritasa_invocations/alembic.py

@invoke.task
def downgrade(
    context: invoke.Context,
    version: str = "base",
) -> None:
    """Downgrade database.

    Use `-v version` to downgrade to the passed version.

    Similar to the `manage.py migrate` command in Django. By default downgrade
    to `base` (similar to `migrate zero`).

    """
    printing.print_success(f"Migrating to version {version}")
    run(context, command=f"downgrade {version}")

load_db_dump(context, file='', env_file_path='.env', reset_db=True)

Reset db and load db dump.

Source code in saritasa_invocations/alembic.py

@invoke.task
def load_db_dump(
    context: invoke.Context,
    file: str = "",
    env_file_path: str = ".env",
    reset_db: bool = True,
) -> None:
    """Reset db and load db dump."""
    if reset_db:
        downgrade(context)
    db.load_db_dump(
        context,
        file=file,
        **_load_local_env_db_settings(context, file=env_file_path),
    )

load_remote_db(context, file='')

Make dump of remote db, download it and apply it.

Source code in saritasa_invocations/alembic.py

@invoke.task
def load_remote_db(
    context: invoke.Context,
    file: str = "",
) -> None:
    """Make dump of remote db, download it and apply it."""
    file = backup_remote_db(context, file=file)
    load_db_dump(context, file=file)

run(context, command)

Execute alembic command.

Source code in saritasa_invocations/alembic.py

@invoke.task
def run(context: invoke.Context, command: str) -> None:
    """Execute alembic command."""
    wait_for_database(context)
    config = _config.Config.from_context(context)
    python.run(
        context,
        command=f"{config.alembic.command} {command}",
    )

upgrade(context, version='head')

Upgrade database.

Use -v version to upgrade to the passed version.

Similar to the manage.py migrate command in Django. By default upgrade to the head - the latest version.

Source code in saritasa_invocations/alembic.py

@invoke.task
def upgrade(
    context: invoke.Context,
    version: str = "head",
) -> None:
    """Upgrade database.

    Use `-v version` to upgrade to the passed version.

    Similar to the `manage.py migrate` command in Django. By default upgrade to
    the `head` - the latest version.

    """
    printing.print_success(f"Migrating to version {version}")
    run(context, command=f"upgrade {version}")

wait_for_database(context)

Ensure that database is up and ready to accept connections.

Function called just once during subsequent calls of alembic commands.

Source code in saritasa_invocations/alembic.py

@invoke.task
def wait_for_database(context: invoke.Context) -> None:
    """Ensure that database is up and ready to accept connections.

    Function called just once during subsequent calls of alembic commands.

    """
    if hasattr(wait_for_database, "_called"):
        return
    docker.up(context)
    printing.print_success("Wait for database connection")
    config = _config.Config.from_context(context)
    with _config.context_override(
        context,
        run={
            "echo": False,
            "hide": "out",
        },
    ) as context:
        for _ in range(config.alembic.connect_attempts - 1):
            try:
                # Doing it manually to avoid loop
                python.run(
                    context,
                    command=f"{config.alembic.command} current",
                )
                wait_for_database._called = True  # type: ignore
                return
            except invoke.UnexpectedExit:  # noqa: PERF203
                time.sleep(1)
                continue

    with _config.context_override(
        context,
        run={
            "echo": True,
            "hide": None,
        },
    ) as context:
        try:
            # Do it one more time but without hiding the terminal output
            python.run(
                context,
                command=f"{config.alembic.command} current",
            )
            wait_for_database._called = True  # type: ignore
        except invoke.UnexpectedExit as error:
            printing.print_error(
                "Failed to connect to db, "
                f"after {config.alembic.connect_attempts} attempts",
            )
            raise invoke.Exit(code=1) from error