DataOps-platform/app/commands/reconcile_dataflow_creates.py

"""Inspect or repair orphaned governed DataFlow create Sagas."""

from __future__ import annotations

import argparse
import json

from app import create_app, db
from app.core.data_flow.create_reconciliation import DataFlowCreateReconciler
from app.core.data_rules.repository import DataRuleRepository


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Reconcile database-owned governed DataFlow create intents"
    )
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument("--apply", action="store_true")
    mode.add_argument("--dry-run", action="store_true")
    parser.add_argument("--limit", type=int, default=100)
    parser.add_argument("--lease-seconds", type=int, default=60)
    args = parser.parse_args()

    app = create_app()
    with app.app_context():
        report = DataFlowCreateReconciler(
            DataRuleRepository(db.session)
        ).run(
            dry_run=not args.apply,
            limit=args.limit,
            lease_seconds=args.lease_seconds,
        )
        print(json.dumps(report, ensure_ascii=False, sort_keys=True))


if __name__ == "__main__":
    main()