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 CORE 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.
  - dataplugins: the same thing for every PLUGIN chain - the applied revision of
                each alembic_version_<plugin> table. The core chain alone cannot
                answer compatibility: plugin schemas move independently, so code
                rolled back to an older plugin revision can sit at an identical
                core head while the dump carries plugin tables it has never heard
                of. Gating on core only let that pass, load, and then fail during
                the plugin upgrade with the data already replaced.

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 _parse_dataplugins(raw):
    """The applied per-plugin revisions, as a {plugin: revision} map.

    Bad or absent JSON becomes an empty map rather than an error: a manifest
    that cannot say what the plugin chains were at is exactly the legacy case
    the restore gate already knows how to treat conservatively, and failing the
    backup here would lose the set over a stamp.
    """
    if not raw:
        return {}
    try:
        value = json.loads(raw)
    except ValueError:
        return {}
    if not isinstance(value, dict):
        return {}
    return {str(k): str(v) for k, v in value.items() if k and v}


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 '',
        'dataplugins': _parse_dataplugins(args.dataplugins),
        '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
    if args.dataplugins is not None:
        manifest['dataplugins'] = _parse_dataplugins(args.dataplugins)
    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('--dataplugins',
                        help='JSON object of {plugin: applied revision}')
    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())