shopdb-flask/scripts/backup/manifest.py

#!/usr/bin/env python3
"""Backup manifest writer.

The manifest is the stamp that turns "we have dumps" into "we can actually
restore." It records, at capture time:
  - takenat   : ISO-8601 UTC timestamp of the run
  - site      : the site tag
  - imagetag  : the app image tag that produced the data
  - schemarev : the core Alembic head revision(s) the running CODE is at, plus
                each plugin chain's heads - the versions a restore target must
                be at least, or restored rows reference columns the live schema
                has not applied yet.
  - datarev   : the Alembic revision actually applied to the DATABASE that was
                dumped (its alembic_version), which is the number a restore must
                gate on. schemarev stamps the code on disk; datarev stamps the
                data in the dump. They diverge when an image is ahead of a DB
                that has not run 'flask db upgrade' yet, and that divergence is
                exactly what a code-to-code check cannot see.

Two modes:
  --out ...      write a fresh manifest for a new set
  --amend ...    patch an existing manifest (used to record ischeckpassed after
                 restic check runs, or to backfill datarev)

ASCII only, no emojis - matches the repo style gate.
"""

import argparse
import json
import os
import sys


def _heads_for(chain_dir):
    """Return the Alembic head revision id(s) under a versions directory.

    We read the revision graph the cheap way: a head is a revision no other
    revision names as its down_revision. Done by text scan so this stays
    dependency-light and works for both the core chain and each plugin chain.
    """
    versions = os.path.join(chain_dir, 'versions')
    if not os.path.isdir(versions):
        return []
    revisions = {}
    down_refs = set()
    for name in os.listdir(versions):
        if not name.endswith('.py'):
            continue
        path = os.path.join(versions, name)
        revid = None
        downid = None
        try:
            with open(path, 'r', encoding='utf-8') as handle:
                for line in handle:
                    stripped = line.strip()
                    if stripped.startswith('revision =') and revid is None:
                        revid = stripped.split('=', 1)[1].strip().strip('"\'')
                    elif stripped.startswith('down_revision ='):
                        value = stripped.split('=', 1)[1].strip()
                        downid = None if value in ('None', 'null') else value.strip('"\'')
        except OSError:
            continue
        if revid:
            revisions[revid] = downid
            if downid:
                down_refs.add(downid)
    return sorted(rev for rev in revisions if rev not in down_refs)


def _collect_schema(migrations_dir, plugins_dir):
    """Core heads plus each plugin's chain heads, keyed for compatibility checks."""
    schema = {'core': _heads_for(migrations_dir), 'plugins': {}}
    if plugins_dir and os.path.isdir(plugins_dir):
        for name in sorted(os.listdir(plugins_dir)):
            chain = os.path.join(plugins_dir, name, 'migrations')
            heads = _heads_for(chain)
            if heads:
                schema['plugins'][name] = heads
    return schema


def write_manifest(args):
    manifest = {
        'takenat': args.takenat,
        'site': args.site,
        'imagetag': args.imagetag,
        'schemarev': _collect_schema(args.migrations, args.plugins),
        'datarev': args.datarev or '',
        'ischeckpassed': False,
    }
    with open(args.out, 'w', encoding='utf-8') as handle:
        json.dump(manifest, handle, indent=2, sort_keys=True)
    return 0


def amend_manifest(args):
    try:
        with open(args.amend, 'r', encoding='utf-8') as handle:
            manifest = json.load(handle)
    except (OSError, ValueError):
        manifest = {}
    if args.ischeckpassed is not None:
        manifest['ischeckpassed'] = args.ischeckpassed.lower() == 'true'
    if args.datarev is not None:
        manifest['datarev'] = args.datarev
    with open(args.amend, 'w', encoding='utf-8') as handle:
        json.dump(manifest, handle, indent=2, sort_keys=True)
    return 0


def main(argv=None):
    parser = argparse.ArgumentParser(description='ShopDB backup manifest writer')
    parser.add_argument('--out')
    parser.add_argument('--amend')
    parser.add_argument('--takenat', default='')
    parser.add_argument('--site', default='shopdb')
    parser.add_argument('--imagetag', default='source')
    parser.add_argument('--migrations', default='/app/migrations')
    parser.add_argument('--plugins', default='/app/plugins')
    parser.add_argument('--datarev')
    parser.add_argument('--ischeckpassed')
    args = parser.parse_args(argv)

    if args.amend:
        return amend_manifest(args)
    if args.out:
        return write_manifest(args)
    parser.error('one of --out or --amend is required')


if __name__ == '__main__':
    sys.exit(main())