tdfs4ds.migration.script¶
Generate a standalone Python script that re-plans and executes a cross-store migration with a baked-in approval map (v0.3.1.150).
Mirrors :mod:tdfs4ds.scheduling.script: the script re-derives a fresh manifest
at runtime (plan_migration) from the baked-in seeds / source /
target_data_domain and then runs migrate_objects with the baked-in
approvals — so the operator can review the emitted .py before executing a
lab→production promotion. Each worker opens its own Teradata context from the
TDFS4DS_TD_* environment via reconnect_teradata (same as the scheduler's
generated script).
module-attribute
¶
_TEMPLATE = '#!/usr/bin/env python\n"""\nAuto-generated by tdfs4ds {version} on {generated_at}.\n\nCross-store MIGRATION script (lab -> production promotion).\nSource: {source_view} -> primary schema (from the environment).\n{n_processes} process(es), {n_objects} object(s), {n_conflicts} conflict(s).\n\nReview the APPROVALS below before running — a `replace` promotes over an existing\nproduction object. Run: python {basename}\n"""\nimport json\n\nSOURCE = json.loads({source_literal})\nSEEDS = json.loads({seeds_literal})\nTARGET_DATA_DOMAIN = {target_domain!r}\nDATA_STRATEGY = {data_strategy!r}\nAPPROVALS = json.loads({approvals_literal})\n# Per-table decisions taken on the reviewed plan (copy the table, or read it in\n# place). Re-applied after re-planning: the object keys are stable.\nDATA_CHOICES = json.loads({data_choices_literal})\n# Target database per kind of object (source_table, intermediate_view,\n# process_view, dataset); empty = the primary store\'s own databases.\nDATABASE_MAP = json.loads({database_map_literal})\n\n\ndef _bootstrap():\n import tdfs4ds\n from tdfs4ds.utils.reconnect import reconnect_teradata\n if not reconnect_teradata():\n raise RuntimeError(\n "reconnect_teradata() failed - set TDFS4DS_TD_HOST/USERNAME/PASSWORD "\n "(and optionally _LOGMECH/_DATABASE) in the environment."\n )\n if {primary_schema!r}:\n tdfs4ds.SCHEMA = {primary_schema!r}\n if {primary_schema_view!r}:\n tdfs4ds.SCHEMA_VIEW = {primary_schema_view!r}\n return tdfs4ds\n\n\ndef main():\n tdfs4ds = _bootstrap()\n plan = tdfs4ds.plan_migration(\n seeds=SEEDS or None, source=SOURCE,\n target_data_domain=TARGET_DATA_DOMAIN, data_strategy=DATA_STRATEGY,\n database_map=DATABASE_MAP or None)\n for key, choice in DATA_CHOICES.items():\n if key in plan.get("objects", {{}}):\n plan["objects"][key]["data_choice"] = choice\n print(tdfs4ds.render_migration_report(plan))\n if plan.get("cycle"):\n raise SystemExit("Circular dependency in the plan - aborting.")\n report = tdfs4ds.migrate_objects(plan, approvals=APPROVALS, progress=print)\n print(json.dumps(report, indent=2, default=str))\n\n\nif __name__ == "__main__":\n main()\n'
The plan's normalised map back in the argument's form ('*' alone = a name).
generate_migration_script(plan: Dict[str, Any], approvals: Optional[Dict[str, str]] = None, *, data_strategy: Optional[str] = None) -> str
Return a standalone .py (string) that re-plans and executes this migration
with the given approvals baked in. Inspect before executing.
write_migration_script(plan: Dict[str, Any], out_path: Optional[str] = None, approvals: Optional[Dict[str, str]] = None, **kw) -> str
Write the generated migration script under tdfs4ds.SCHEDULING_SCRIPT_DIR
(or out_path); return its path.