#!/usr/bin/env python3 """Migrate DynamoDB table data from acdl-* → nova-* (REQ-163, P4). The Nova rebrand (v1.15) renames the platform DynamoDB tables: - ``acdl-contracts`` → ``nova-contracts`` - ``acdl-change-requests`` → ``nova-change-requests`` DynamoDB table names are immutable post-creation, so the migration is a **scan + copy**: every item in the old table is written to the new table (preserving the full item shape — PK, SK, and all attributes). Row counts are verified to match post-copy. The old tables are **kept** until the operator verifies the copy; deletion is a manual post-verification step documented in ``docs/NOVA_AWS_MIGRATION.md`` (runbook). Design: - **Dry-run by default.** Prints the planned copy operations + counts without touching AWS. Pass ``--apply`` to execute the copy. - **Idempotent.** Re-running against an already-migrated item is a no-op (``PutItem`` overwrites in place; the copy is re-run but the row counts still match). The script does NOT delete the old tables (deletion is a manual runbook step). - **Item-mapping logic is pure + unit-tested** (see ``tests/test_migrate_dynamodb_data.py``); the AWS I/O is thin boto3 glue around ``map_item()`` + ``scan_all()``. - **boto3 lazy import.** The module is importable + unit-testable without AWS credentials (the client is constructed inside ``run()``). Usage: python3 scripts/migrate_dynamodb_data.py # dry-run (default) python3 scripts/migrate_dynamodb_data.py --apply # execute the copy python3 scripts/migrate_dynamodb_data.py --region us-east-1 --apply python3 scripts/migrate_dynamodb_data.py --table contracts --apply python3 scripts/migrate_dynamodb_data.py --source acdl-contracts --dest nova-contracts --apply Pre-requisites (live AWS, documented in the runbook): - The nova-* destination tables must already exist (created via ``terraform/platform/main.tf``). - AWS credentials in env with scan+PutItem on both old + new tables. """ from __future__ import annotations import argparse import copy import sys from typing import Dict, List, Optional, Tuple try: import boto3 except ImportError: # pragma: no cover - boto3 is a test dep boto3 = None # type: ignore # --------------------------------------------------------------------------- # Default table-pair mapping (REQ-163) # --------------------------------------------------------------------------- DEFAULT_TABLE_PAIRS: List[Tuple[str, str]] = [ ("acdl-contracts", "nova-contracts"), ("acdl-change-requests", "nova-change-requests"), ] # --------------------------------------------------------------------------- # Pure item-mapping logic (unit-tested) # --------------------------------------------------------------------------- def map_item(item: Dict) -> Dict: """Return a copy of a DynamoDB item suitable for PutItem into the new table. DynamoDB items returned by ``scan``/``get_item`` are in the typed-attribute shape (``{"attr": {"S": "value"}, ...}``). The copy is identity-preserving: the item is written verbatim to the destination table so the PK/SK + every attribute land identically. No key-rewrite is needed because the old + new tables share the same key schema (PK ``consumerRepo``, SK ``contractId#submittedAt`` for contracts; PK ``changeRequestId``, SK ``submittedAt`` for change-requests). The mapping is a deep copy so callers can mutate the result without aliasing the scanned item (DynamoDB items nest typed-attribute dicts, e.g. ``{"attr": {"S": "value"}}``). ``map_item`` is pure + side-effect-free. Examples: >>> map_item({"consumerRepo": {"S": "acdl/c"}, "k": {"N": "1"}}) {'consumerRepo': {'S': 'acdl/c'}, 'k': {'N': '1'}} >>> map_item({}) == {} True """ return copy.deepcopy(item) def table_pair_for(name: str, pairs: Optional[List[Tuple[str, str]]] = None) -> Tuple[str, str]: """Resolve a logical table name (``contracts`` / ``change-requests``) or a literal source-table name to its ``(source, dest)`` pair. Examples: >>> table_pair_for("contracts") ('acdl-contracts', 'nova-contracts') >>> table_pair_for("change-requests") ('acdl-change-requests', 'nova-change-requests') >>> table_pair_for("acdl-contracts") ('acdl-contracts', 'nova-contracts') >>> table_pair_for("nova-contracts") ('nova-contracts', 'nova-contracts') """ table = pairs if pairs is not None else DEFAULT_TABLE_PAIRS aliases = { "contracts": ("acdl-contracts", "nova-contracts"), "change-requests": ("acdl-change-requests", "nova-change-requests"), } if name in aliases: return aliases[name] for src, dst in table: if name == src: return (src, dst) if name == dst: return (src, dst) raise ValueError( f"unknown table {name!r}; expected one of: contracts, change-requests, " f"or a literal source name from {table!r}" ) # --------------------------------------------------------------------------- # Thin AWS I/O glue (constructed lazily inside run) # --------------------------------------------------------------------------- def scan_all(client, table_name: str) -> List[Dict]: """Scan every item in ``table_name`` (paginates through all segments). Returns the full list of items (typed-attribute shape). Uses ``table.scan()`` with pagination on ``LastEvaluatedKey``. """ items: List[Dict] = [] last_key: Optional[Dict] = None while True: kwargs: Dict = {"TableName": table_name} if last_key is not None: kwargs["ExclusiveStartKey"] = last_key resp = client.scan(**kwargs) items.extend(resp.get("Items", [])) last_key = resp.get("LastEvaluatedKey") if not last_key: break return items def copy_items(client, source_table: str, dest_table: str, items: List[Dict]) -> int: """PutItem every mapped item into ``dest_table``. Returns the count written.""" written = 0 for item in items: client.put_item(TableName=dest_table, Item=map_item(item)) written += 1 return written def count_items(client, table_name: str) -> int: """Return the approximate item count via ``DescribeTable``. Uses ``Table.ItemCount`` (updated ~6hourly by AWS) for a fast count; for exact verification prefer ``len(scan_all(...))`` (the runbook documents both — scan is the source of truth for row-count verification). """ resp = client.describe_table(TableName=table_name) return int(resp["Table"].get("ItemCount", 0)) # --------------------------------------------------------------------------- # Driver # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> int: pairs = DEFAULT_TABLE_PAIRS if args.source and args.dest: pairs = [(args.source, args.dest)] elif args.table: pairs = [table_pair_for(args.table)] region = args.region if boto3 is None: print("FAIL: boto3 is not installed (pip install boto3)", file=sys.stderr) return 2 client = boto3.client("dynamodb", region_name=region) mode = "APPLY" if args.apply else "DRY-RUN" overall_rc = 0 for source, dest in pairs: print(f"\n=== {mode}: {source} → {dest} (region {region}) ===") try: client.describe_table(TableName=source) except Exception as e: print(f" FAIL: source table {source!r} not describable: " f"{type(e).__name__}: {e}", file=sys.stderr) overall_rc = 1 continue try: client.describe_table(TableName=dest) except Exception as e: print(f" FAIL: dest table {dest!r} not describable (create it via " f"terraform first): {type(e).__name__}: {e}", file=sys.stderr) overall_rc = 1 continue items = scan_all(client, source) src_count = len(items) print(f" scanned {src_count} item(s) from {source}") if not args.apply: print(f" [dry-run] would PutItem {src_count} item(s) into {dest}") print(f" [dry-run] would verify {dest} row count == {src_count}") print(f" [dry-run] old table {source} is NOT deleted (manual runbook step)") continue written = copy_items(client, source, dest, items) print(f" copied {written} item(s) → {dest}") # Verify by re-scanning the destination (source of truth, not DescribeTable). dest_items = scan_all(client, dest) dest_count = len(dest_items) if dest_count != src_count: print(f" WARNING: row-count mismatch — source={src_count}, " f"dest={dest_count}. Investigate before deleting {source}.", file=sys.stderr) overall_rc = 1 else: print(f" VERIFIED: {dest} row count ({dest_count}) == source ({src_count})") print(f" Old table {source} is KEPT. Delete it manually only after " f"verifying consumers read from {dest} (runbook step).") if overall_rc == 0: print(f"\n=== {mode} complete ({len(pairs)} pair(s)) ===") else: print(f"\n=== {mode} complete with FAILURES ===", file=sys.stderr) return overall_rc def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( description="Migrate DynamoDB data acdl-* → nova-* (REQ-163, P4).", ) p.add_argument("--apply", action="store_true", help="Execute the copy (default: dry-run, no AWS writes).") p.add_argument("--region", default="us-east-1", help="AWS region (default: us-east-1).") p.add_argument("--table", default=None, help="Migrate a single logical table: 'contracts' or " "'change-requests' (default: both).") p.add_argument("--source", default=None, help="Override the source table name (paired with --dest).") p.add_argument("--dest", default=None, help="Override the destination table name (paired with --source).") return p def main(argv: Optional[List[str]] = None) -> int: args = build_parser().parse_args(argv) return run(args) if __name__ == "__main__": sys.exit(main())