1from __future__ import annotations
  2
  3import os
  4import time
  5from pathlib import Path
  6
  7import click
  8
  9from .core import DatabaseBackups, get_git_branch
 10
 11
 12@click.group("backups")
 13def cli() -> None:
 14    """Local database backups"""
 15    pass
 16
 17
 18@cli.command("list")
 19@click.option(
 20    "--branch",
 21    "branch",
 22    is_flag=False,
 23    flag_value="__current__",
 24    default=None,
 25    help="Filter by branch (defaults to current branch if flag used without value)",
 26)
 27def list_backups(branch: str | None) -> None:
 28    """List database backups"""
 29    backups_handler = DatabaseBackups()
 30    backups = backups_handler.find_backups()
 31
 32    # Resolve branch filter
 33    if branch == "__current__":
 34        branch = get_git_branch()
 35
 36    # Filter by branch if specified
 37    if branch:
 38        backups = [b for b in backups if b.metadata.get("git_branch") == branch]
 39
 40    if not backups:
 41        if branch:
 42            click.secho(f"No backups found for branch '{branch}'", fg="yellow")
 43        else:
 44            click.secho("No backups found", fg="yellow")
 45        return
 46
 47    # Calculate column widths
 48    name_width = max(len(b.name) for b in backups)
 49    source_width = max(len(b.metadata.get("source") or "-") for b in backups)
 50
 51    # Print header
 52    click.secho(
 53        f"{'NAME':<{name_width}}  {'SOURCE':<{source_width}}  {'SIZE':<10}  BRANCH",
 54        dim=True,
 55    )
 56
 57    # Print rows
 58    for backup in backups:
 59        backup_file = backup.path / "default.backup"
 60        if backup_file.exists():
 61            size = os.path.getsize(backup_file)
 62            size_str = f"{size / 1024 / 1024:.2f} MB"
 63        else:
 64            size_str = "-"
 65        metadata = backup.metadata
 66        source = metadata.get("source") or "-"
 67        git_branch = metadata.get("git_branch") or "-"
 68
 69        click.echo(
 70            f"{backup.name:<{name_width}}  {source:<{source_width}}  {size_str:<10}  {git_branch}"
 71        )
 72
 73
 74@cli.command("create")
 75@click.option("--pg-dump", default="pg_dump", envvar="PG_DUMP")
 76@click.argument("backup_name", default="")
 77def create_backup(backup_name: str, pg_dump: str) -> None:
 78    """Create a database backup"""
 79    backups_handler = DatabaseBackups()
 80
 81    if not backup_name:
 82        backup_name = time.strftime("%Y%m%d_%H%M%S")
 83
 84    try:
 85        backup_dir = backups_handler.create(
 86            backup_name,
 87            source="manual",
 88            pg_dump=pg_dump,
 89        )
 90    except Exception as e:
 91        click.secho(str(e), fg="red")
 92        exit(1)
 93
 94    click.secho(f"Backup created in {backup_dir.relative_to(Path.cwd())}", fg="green")
 95
 96
 97@cli.command("restore")
 98@click.option("--latest", is_flag=True)
 99@click.option("--pg-restore", default="pg_restore", envvar="PG_RESTORE")
100@click.argument("backup_name", default="")
101def restore_backup(backup_name: str, latest: bool, pg_restore: str) -> None:
102    """Restore a database backup"""
103    backups_handler = DatabaseBackups()
104
105    if backup_name and latest:
106        raise click.UsageError("Only one of --latest or backup_name is allowed")
107
108    if not backup_name and not latest:
109        raise click.UsageError("Backup name or --latest is required")
110
111    if not backup_name and latest:
112        backup_name = backups_handler.find_backups()[0].name
113
114    click.secho(f"Restoring backup {backup_name}...", bold=True)
115
116    try:
117        backups_handler.restore(
118            backup_name,
119            pg_restore=pg_restore,
120        )
121    except Exception as e:
122        click.secho(str(e), fg="red")
123        exit(1)
124    click.echo(f"Backup {backup_name} restored successfully.")
125
126
127@cli.command("delete")
128@click.argument("backup_name")
129def delete_backup(backup_name: str) -> None:
130    """Delete a database backup"""
131    backups_handler = DatabaseBackups()
132    try:
133        backups_handler.delete(backup_name)
134    except Exception as e:
135        click.secho(str(e), fg="red")
136        return
137    click.secho(f"Backup {backup_name} deleted", fg="green")
138
139
140@cli.command("clear")
141@click.confirmation_option(prompt="Are you sure you want to delete all backups?")
142def clear_backups() -> None:
143    """Clear all database backups"""
144    backups_handler = DatabaseBackups()
145    backups = backups_handler.find_backups()
146    for backup in backups:
147        backup.delete()
148    click.secho("All backups deleted", fg="green")