Desenhar e validar uma migração de dados

"""Original offline DR exercise. Declared record equivalence is not migration approval."""
import json
import re
import sys
from collections import Counter
from datetime import datetime, timezone


def fields(value, expected):
    if type(value) is not dict or set(value) != set(expected):
        raise ValueError('Unexpected or missing fields')


def label(value):
    if type(value) is not str or not value or value != value.strip():
        raise ValueError('Expected a nonempty label without surrounding whitespace')
    return value


def amount_cents(value):
    # Exact integer arithmetic. No binary float conversion or implicit rounding.
    if type(value) is not str or not re.fullmatch(r'-?(0|[1-9][0-9]{0,15})(\.[0-9]{1,2})?', value):
        raise ValueError('Expected a bounded decimal string with at most two decimal places')
    negative = value.startswith('-')
    whole, _, fraction = value.lstrip('-').partition('.')
    cents = int(whole) * 100 + int(fraction.ljust(2, '0'))
    return -cents if negative else cents


def instant(value):
    if type(value) is not str or not re.fullmatch(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(Z|[+-]\d{2}:\d{2})', value):
        raise ValueError('Expected whole-second ISO timestamp with explicit offset')
    if value.endswith('-00:00'):
        raise ValueError('Unknown local offset is not an established instant')
    if not value.endswith('Z'):
        hours, minutes = map(int, value[-5:].split(':'))
        if hours > 23 or minutes > 59:
            raise ValueError('Invalid offset')
    parsed = datetime.fromisoformat(value.replace('Z', '+00:00'))
    return parsed.astimezone(timezone.utc).isoformat()


def dataset(value):
    fields(value, ['snapshot', 'complete', 'rows'])
    label(value['snapshot'])
    if type(value['complete']) is not bool or type(value['rows']) is not list:
        raise ValueError('Expected explicit completeness and row list')
    rows = []
    for row in value['rows']:
        fields(row, ['id', 'currency', 'amount', 'at', 'reference'])
        rid = label(row['id'])
        if type(row['currency']) is not str or not re.fullmatch(r'[A-Z]{3}', row['currency']):
            raise ValueError('Currency must be a three-letter code; no conversion is performed')
        if row['reference'] is not None and type(row['reference']) is not str:
            raise ValueError('Reference must be string or null')
        rows.append({'id': rid, 'currency': row['currency'], 'cents': amount_cents(row['amount']),
                     'instant': instant(row['at']), 'reference': row['reference']})
    return rows


def reconcile(value):
    fields(value, ['source', 'target'])
    left, right = dataset(value['source']), dataset(value['target'])
    counts = [Counter(row['id'] for row in rows) for rows in [left, right]]
    duplicates = {'source': sorted(k for k, n in counts[0].items() if n > 1),
                  'target': sorted(k for k, n in counts[1].items() if n > 1)}
    maps = [{row['id']: row for row in rows if count[row['id']] == 1}
            for rows, count in zip([left, right], counts)]
    source_ids, target_ids = set(counts[0]), set(counts[1])
    changed = []
    for rid in sorted(set(maps[0]) & set(maps[1])):
        differing = [k for k in ['currency', 'cents', 'instant', 'reference'] if maps[0][rid][k] != maps[1][rid][k]]
        if differing:
            changed.append({'id': rid, 'fields': differing})
    totals = []
    for rows in [left, right]:
        result = {}
        for row in rows:
            result[row['currency']] = result.get(row['currency'], 0) + row['cents']
        totals.append(dict(sorted(result.items())))
    missing, extra = sorted(source_ids-target_ids), sorted(target_ids-source_ids)
    blockers = []
    if value['source']['snapshot'] != value['target']['snapshot']:
        blockers.append('different-declared-snapshots')
    if not value['source']['complete'] or not value['target']['complete']:
        blockers.append('incomplete-declared-scope')
    if any(duplicates.values()):
        blockers.append('nonunique-keys')
    if missing or extra or changed:
        blockers.append('record-differences')
    return {'equivalentUnderDeclaredContract': not blockers, 'blockers': blockers,
            'missing': missing, 'unexpected': extra, 'changed': changed, 'duplicates': duplicates,
            'rowCounts': [len(left), len(right)], 'totalsCentsByCurrency': totals,
            'snapshotAuthenticityVerified': False, 'accessControlsVerified': False,
            'productionCutoverApproved': False}


if __name__ == '__main__':
    try:
        result = reconcile(json.load(sys.stdin))
        print(json.dumps(result, sort_keys=True))
    except (ValueError, TypeError, OverflowError) as error:
        print(json.dumps({'error': str(error)}))
        sys.exit(2)

Tópicos relacionados: Ingestão e reprocessamento · Modelação e armazenamento · Observabilidade de pipelines

Referência: Database migration concepts and principles · Current linked standard guide (document title v4.2); edition date unconfirmed (2026-09-30 inspection)