kamihi.cli.commands.db

Database management module for Kamihi CLI.

License

MIT

Functions:

Name	Description
downgrade	Downgrade the database to an earlier version.
main	Database management commands for Kamihi CLI.
migrate	Run database migrations.
revision_callback	Ensure the revision value is valid.
upgrade	Upgrade the database to a later version.

downgrade

downgrade(
    ctx: Context,
    revision: Annotated[
        str,
        Option(
            --revision,
            -r,
            help="The revision to downgrade to.",
            show_default=-1,
            callback=revision_callback,
        ),
    ] = "-1",
) -> None

Downgrade the database to an earlier version.

Source code in src/kamihi/cli/commands/db.py

@app.command("downgrade")
def downgrade(
    ctx: typer.Context,
    revision: Annotated[
        str,
        typer.Option(
            "--revision", "-r", help="The revision to downgrade to.", show_default="-1", callback=revision_callback
        ),
    ] = "-1",
) -> None:
    """Downgrade the database to an earlier version."""
    command.downgrade(ctx.obj.alembic_cfg, revision)
    logger.bind(revision=revision).success("Downgraded")

main

main(ctx: Context) -> None

Database management commands for Kamihi CLI.

Source code in src/kamihi/cli/commands/db.py

@app.callback()
def main(ctx: typer.Context) -> None:
    """Database management commands for Kamihi CLI."""
    settings = get_settings()

    import_models(ctx.obj.cwd / "models")

    ctx.obj.alembic_cfg = Config(toml_file=ctx.obj.cwd / "pyproject.toml")
    ctx.obj.alembic_cfg.set_main_option("sqlalchemy.url", settings.db.url)

    if not (ctx.obj.cwd / "migrations").exists():
        logger.error("No migrations directory found. Please run 'kamihi init' first.")
        raise typer.Exit(code=1)

migrate

migrate(ctx: Context) -> None

Run database migrations.

Source code in src/kamihi/cli/commands/db.py

@app.command("migrate")
def migrate(ctx: typer.Context) -> None:
    """Run database migrations."""
    with contextlib.redirect_stdout(StreamToLogger(logger, "DEBUG")):
        res = command.revision(ctx.obj.alembic_cfg, autogenerate=True, message="auto migration")
    logger.bind(revision=res.revision).success("Migrated")

revision_callback

revision_callback(ctx: Context, value: str) -> str

Ensure the revision value is valid.

Parameters:

Name	Type	Description	Default
ctx	Context	The Typer context.	required
value	str	The revision value.	required

Returns:

Name	Type	Description
str	str	The validated revision value.

Source code in src/kamihi/cli/commands/db.py

def revision_callback(ctx: typer.Context, value: str) -> str:
    """
    Ensure the revision value is valid.

    Args:
        ctx (typer.Context): The Typer context.
        value (str): The revision value.

    Returns:
        str: The validated revision value.

    """
    if not value or not isinstance(value, str):
        raise typer.BadParameter("Invalid revision value")
    script = ScriptDirectory.from_config(ctx.obj.alembic_cfg)
    value = script.as_revision_number(value)
    if value is None:
        raise typer.BadParameter("Revision not found")
    return value

upgrade

upgrade(
    ctx: Context,
    revision: Annotated[
        str,
        Option(
            --revision,
            -r,
            help="The revision to upgrade to.",
            show_default=head,
            callback=revision_callback,
        ),
    ] = "head",
) -> None

Upgrade the database to a later version.

Source code in src/kamihi/cli/commands/db.py

@app.command("upgrade")
def upgrade(
    ctx: typer.Context,
    revision: Annotated[
        str,
        typer.Option(
            "--revision", "-r", help="The revision to upgrade to.", show_default="head", callback=revision_callback
        ),
    ] = "head",
) -> None:
    """Upgrade the database to a later version."""
    command.upgrade(ctx.obj.alembic_cfg, revision)
    logger.bind(revision=revision).success("Upgraded")