38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
"""Command-line entry point for database migrations."""
|
|
|
|
import argparse
|
|
|
|
from core.migrations import MigrationError, migration_status, upgrade
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description="Manage database migrations")
|
|
parser.add_argument("command", choices=("upgrade", "status"))
|
|
args = parser.parse_args(argv)
|
|
|
|
try:
|
|
if args.command == "upgrade":
|
|
applied = upgrade()
|
|
for migration in applied:
|
|
print(f"applied {migration.namespace}:{migration.version} {migration.name}")
|
|
if not applied:
|
|
print("database is up to date")
|
|
return 0
|
|
|
|
records = migration_status()
|
|
if not records:
|
|
print("no migrations found")
|
|
for record in records:
|
|
name = record["name"] or "<source missing>"
|
|
print(
|
|
f"{record['state']:<8} "
|
|
f"{record['namespace']}:{record['version']} {name}"
|
|
)
|
|
return 1 if any(row["state"] == "changed" for row in records) else 0
|
|
except MigrationError as error:
|
|
parser.exit(1, f"migration error: {error}\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|