45 lines
1.8 KiB
Python
45 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections.abc import Sequence
|
|
|
|
from backend.bootstrap.settings import Settings
|
|
from backend.database import MIGRATIONS, Database, MigrationRunner
|
|
from backend.database.repositories import DatabaseStatusRepository
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Manage the rebuilt application database")
|
|
subcommands = parser.add_subparsers(dest="command", required=True)
|
|
subcommands.add_parser("status", help="Show database availability and schema version")
|
|
subcommands.add_parser("upgrade", help="Apply all pending migrations")
|
|
downgrade = subcommands.add_parser("downgrade", help="Roll back to a schema version")
|
|
downgrade.add_argument("--target", type=int, required=True)
|
|
downgrade.add_argument("--confirm-downgrade", action="store_true")
|
|
return parser
|
|
|
|
|
|
def main(arguments: Sequence[str] | None = None) -> int:
|
|
parsed = build_parser().parse_args(arguments)
|
|
settings = Settings.from_environment()
|
|
database = Database(settings.database_path)
|
|
runner = MigrationRunner(database)
|
|
|
|
if parsed.command == "status":
|
|
status = DatabaseStatusRepository(database).get()
|
|
print(f"available={str(status.available).lower()} schema_version={status.schema_version}")
|
|
return 0
|
|
if parsed.command == "upgrade":
|
|
applied = runner.upgrade(MIGRATIONS)
|
|
print("applied=" + (",".join(f"{version:04d}" for version in applied) or "none"))
|
|
return 0
|
|
if not parsed.confirm_downgrade:
|
|
raise SystemExit("downgrade requires --confirm-downgrade")
|
|
rolled_back = runner.downgrade(MIGRATIONS, parsed.target)
|
|
print("rolled_back=" + (",".join(f"{version:04d}" for version in rolled_back) or "none"))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|