DataOps-platform/app/commands/reconcile_workflow_versions.py

from __future__ import annotations

from sqlalchemy import text

from app import create_app, db
from app.core.data_factory.n8n_client import N8nClient, N8nClientError


def main() -> None:
    app = create_app()
    with app.app_context():
        rows = db.session.execute(
            text(
                "SELECT dataflow_uid::text, environment, version_no, n8n_workflow_id "
                "FROM public.dataflow_workflow_versions WHERE status = 'active' "
                "ORDER BY dataflow_uid, environment"
            )
        )
        client = N8nClient()
        mismatch_count = 0
        for uid, environment, version_no, workflow_id in rows:
            try:
                workflow = client.get_workflow(workflow_id)
                remote_active = bool(workflow.get("active"))
                state = "match" if remote_active else "mismatch_remote_inactive"
            except N8nClientError as exc:
                state = f"unavailable:{exc.message}"
            if state != "match":
                mismatch_count += 1
            print(f"{uid}\t{environment}\tv{version_no}\t{workflow_id}\t{state}")
        print(f"mismatches={mismatch_count}; repair_performed=false")


if __name__ == "__main__":
    main()