backup-docker-to-local/src/baudolo/backup/app.py

"""Back up every Docker volume of a host into a timestamped generation."""

from __future__ import annotations

import os
from contextlib import ExitStack
from datetime import datetime

from .cli import parse_args
from .compose import handle_docker_compose_services
from .docker import (
    change_containers_status,
    containers_using_volume,
    docker_volume_names,
    filter_stoppable,
)
from .dumps import backup_dumps_for_volume, load_databases_df
from .layout import (
    create_version_directory,
    create_volume_directory,
    get_machine_id,
    stamp_directory,
)
from .policy import requires_stop, volume_is_fully_ignored
from .snapshot import snapshot_source, volume_snapshot
from .volume import backup_volume, inspect_backing


def main() -> int:
    args = parse_args()

    machine_id = get_machine_id()
    # Local wall clock on purpose: generations sort by this name, and UTC would
    # order new ones before the existing ones wherever the offset is positive.
    backup_time = datetime.now().strftime("%Y%m%d%H%M%S")  # noqa: DTZ005

    versions_dir = os.path.join(args.backups_dir, machine_id, args.repo_name)
    version_dir = create_version_directory(versions_dir, backup_time)

    databases_df = None if args.only_files else load_databases_df(args.databases_csv)

    print("💾 Start volume backups...", flush=True)

    with ExitStack() as stack:
        resolve_source = None
        if args.snapshot:
            resolve_source = stack.enter_context(
                volume_snapshot(args.snapshot, args.snapshot_subject, backup_time)
            )

        for volume_name in docker_volume_names():
            print(f"Start backup routine for volume: {volume_name}", flush=True)

            if volume_name in args.volumes_no_backup_required:
                print(
                    f"Skipping volume '{volume_name}' entirely (declared no-backup).",
                    flush=True,
                )
                continue

            containers = containers_using_volume(volume_name)

            if volume_is_fully_ignored(containers, args.images_no_backup_required):
                print(
                    f"Skipping volume '{volume_name}' entirely (all linked containers are ignored).",
                    flush=True,
                )
                continue

            vol_dir = create_volume_directory(version_dir, volume_name)

            found_db = dumped_any = False
            if not args.only_files:
                found_db, dumped_any = backup_dumps_for_volume(
                    containers=containers,
                    vol_dir=vol_dir,
                    databases_df=databases_df,
                    database_containers=args.database_containers,
                )

            if args.only_sql and found_db:
                if not dumped_any:
                    print(
                        f"WARNING: only-sql requested but no DB dump was produced for DB volume '{volume_name}'. "
                        "Falling back to file backup.",
                        flush=True,
                    )
                else:
                    continue

            backing = inspect_backing(volume_name)
            live_source = backing.source

            def copy(
                *,
                authoritative: bool,
                source: str = live_source,
                volume: str = volume_name,
                target: str = vol_dir,
            ) -> None:
                backup_volume(
                    versions_dir,
                    volume,
                    target,
                    authoritative=authoritative,
                    source=source,
                )

            if resolve_source is not None:
                source, reason = snapshot_source(
                    resolve_source, backing, args.snapshot_subject
                )
                if source is not None:
                    copy(authoritative=True, source=source)
                else:
                    print(
                        f"WARNING: volume '{volume_name}' is not in the snapshot "
                        f"({reason}); copying it live instead.",
                        flush=True,
                    )
                    copy(authoritative=False)
                continue

            copy(authoritative=False)
            if requires_stop(containers, args.images_no_stop_required):
                stoppable = filter_stoppable(containers)
                change_containers_status(stoppable, "stop")
                copy(authoritative=True)
                if not args.shutdown:
                    change_containers_status(stoppable, "start")

    stamp_directory(version_dir)
    print("Finished volume backups.", flush=True)

    print("Handling Docker Compose services...", flush=True)
    handle_docker_compose_services(args.compose_dir, args.hard_restart_projects)

    return 0