feat: adiciona ingestão de DAGs de ingestão para empenhos, faturas, cronogramas e terceirizados

Author: luanaa2005

airflow_lappis/dags/data_ingest/compras_gov/mir/contratos_cronograma_mir_ingest_dag.py

@@ -0,0 +1,60 @@
+import logging
+from airflow.decorators import dag, task
+from datetime import datetime, timedelta
+from schedule_loader import get_dynamic_schedule
+from cliente_contratos import ClienteContratos
+from cliente_postgres import ClientPostgresDB
+from postgres_helpers import get_postgres_conn
+
+
+@dag(
+    dag_id="contratos_cronograma_mir_ingest_dag",
+    schedule_interval=get_dynamic_schedule("contratos_cronograma_mir_ingest_dag"),
+    start_date=datetime(2023, 1, 1),
+    catchup=False,
+    default_args={
+        "owner": "Luana",
+        "retries": 1,
+        "retry_delay": timedelta(minutes=5),
+    },
+    tags=["cronograma_api", "compras_gov", "MIR"],
+)
+def api_cronograma_mir_dag() -> None:
+    """DAG para buscar e armazenar cronograma no PostgreSQL do MIR."""
+
+    @task
+    def fetch_cronograma() -> None:
+        logging.info("[contratos_cronograma_mir_ingest_dag.py] Starting fetch_cronograma task")
+        api = ClienteContratos()
+        postgres_conn_str = get_postgres_conn("postgres_mir")  
+        db = ClientPostgresDB(postgres_conn_str)
+        contratos_ids = db.get_contratos_ids()
+
+        for contrato_id in contratos_ids:
+            try:
+                logging.info(
+                    f"[contratos_cronograma_mir_ingest_dag.py] Fetching cronograma for contrato ID: "
+                    f"{contrato_id}"
+                )
+                cronograma = api.get_cronograma_by_contrato_id(str(contrato_id))
+
+                if cronograma:
+                    for fatura in cronograma:
+                        fatura["dt_ingest"] = datetime.now().isoformat()
+
+                db.insert_data(
+                    cronograma,
+                    "cronograma",
+                    conflict_fields=["id"],
+                    primary_key=["id"],
+                    schema="compras_gov",
+                )
+            except Exception as e:
+                logging.error(
+                    f"[contratos_cronograma_mir_ingest_dag.py] Error for contrato ID {contrato_id}: {e}"
+                )
+
+    fetch_cronograma()
+
+
+dag_instance = api_cronograma_mir_dag()

airflow_lappis/dags/data_ingest/compras_gov/mir/contratos_empenhos_mir_ingest_dag.py

@@ -0,0 +1,60 @@
+import logging
+from airflow.decorators import dag, task
+from datetime import datetime, timedelta
+from schedule_loader import get_dynamic_schedule
+from cliente_contratos import ClienteContratos
+from cliente_postgres import ClientPostgresDB
+from postgres_helpers import get_postgres_conn
+
+
+@dag(
+    dag_id="contratos_empenhos_mir_ingest_dag",
+    schedule_interval=get_dynamic_schedule("contratos_empenhos_mir_ingest_dag"),
+    start_date=datetime(2023, 1, 1),
+    catchup=False,
+    default_args={
+        "owner": "Luana",
+        "retries": 1,
+        "retry_delay": timedelta(minutes=5),
+    },
+    tags=["empenhos_api", "compras_gov", "MIR"],
+)
+def api_empenhos_mir_dag() -> None:
+    """DAG para buscar e armazenar empenhos no PostgreSQL do MIR."""
+
+    @task
+    def fetch_empenhos() -> None:
+        logging.info("[contratos_empenhos_mir_ingest_dag.py] Starting fetch_empenhos task")
+        api = ClienteContratos()
+        postgres_conn_str = get_postgres_conn("postgres_mir")  
+        db = ClientPostgresDB(postgres_conn_str)
+        contratos_ids = db.get_contratos_ids()
+
+        for contrato_id in contratos_ids:
+            try:
+                logging.info(
+                    f"[contratos_empenhos_mir_ingest_dag.py] Fetching empenhos for contrato ID: "
+                    f"{contrato_id}"
+                )
+                empenhos = api.get_empenhos_by_contrato_id(str(contrato_id))
+
+                if empenhos:
+                    for fatura in empenhos:
+                        fatura["dt_ingest"] = datetime.now().isoformat()
+
+                db.insert_data(
+                    empenhos,
+                    "empenhos",
+                    conflict_fields=["id"],
+                    primary_key=["id"],
+                    schema="compras_gov",
+                )
+            except Exception as e:
+                logging.error(
+                    f"[contratos_empenhos_mir_ingest_dag.py] Error for contrato ID {contrato_id}: {e}"
+                )
+
+    fetch_empenhos()
+
+
+dag_instance = api_empenhos_mir_dag()

airflow_lappis/dags/data_ingest/compras_gov/mir/contratos_faturas_mir_ingest_dag.py

@@ -0,0 +1,60 @@
+import logging
+from airflow.decorators import dag, task
+from datetime import datetime, timedelta
+from schedule_loader import get_dynamic_schedule
+from cliente_contratos import ClienteContratos
+from cliente_postgres import ClientPostgresDB
+from postgres_helpers import get_postgres_conn
+
+
+@dag(
+    dag_id="contratos_faturas_mir_ingest_dag",
+    schedule_interval=get_dynamic_schedule("contratos_faturas_mir_ingest_dag"),
+    start_date=datetime(2023, 1, 1),
+    catchup=False,
+    default_args={
+        "owner": "Luana",
+        "retries": 1,
+        "retry_delay": timedelta(minutes=5),
+    },
+    tags=["faturas_api", "compras_gov", "MIR"],
+)
+def api_faturas_mir_dag() -> None:
+    """DAG para buscar e armazenar faturas no PostgreSQL do MIR."""
+
+    @task
+    def fetch_faturas() -> None:
+        logging.info("[contratos_faturas_mir_ingest_dag.py] Starting fetch_faturas task")
+        api = ClienteContratos()
+        postgres_conn_str = get_postgres_conn("postgres_mir")  
+        db = ClientPostgresDB(postgres_conn_str)
+        contratos_ids = db.get_contratos_ids()
+
+        for contrato_id in contratos_ids:
+            try:
+                logging.info(
+                    f"[contratos_faturas_mir_ingest_dag.py] Fetching faturas for contrato ID: "
+                    f"{contrato_id}"
+                )
+                faturas = api.get_faturas_by_contrato_id(str(contrato_id))
+
+                if faturas:
+                    for fatura in faturas:
+                        fatura["dt_ingest"] = datetime.now().isoformat()
+
+                db.insert_data(
+                    faturas,
+                    "faturas",
+                    conflict_fields=["id"],
+                    primary_key=["id"],
+                    schema="compras_gov",
+                )
+            except Exception as e:
+                logging.error(
+                    f"[contratos_faturas_mir_ingest_dag.py] Error for contrato ID {contrato_id}: {e}"
+                )
+
+    fetch_faturas()
+
+
+dag_instance = api_faturas_mir_dag()

airflow_lappis/dags/data_ingest/compras_gov/mir/contratos_terceirizados_mir_ingest_dag.py

@@ -0,0 +1,60 @@
+import logging
+from airflow.decorators import dag, task
+from datetime import datetime, timedelta
+from schedule_loader import get_dynamic_schedule
+from cliente_contratos import ClienteContratos
+from cliente_postgres import ClientPostgresDB
+from postgres_helpers import get_postgres_conn
+
+
+@dag(
+    dag_id="contratos_terceirizados_mir_ingest_dag",
+    schedule_interval=get_dynamic_schedule("contratos_terceirizados_mir_ingest_dag"),
+    start_date=datetime(2023, 1, 1),
+    catchup=False,
+    default_args={
+        "owner": "Luana",
+        "retries": 1,
+        "retry_delay": timedelta(minutes=5),
+    },
+    tags=["terceirizados_api", "compras_gov", "MIR"],
+)
+def api_terceirizados_mir_dag() -> None:
+    """DAG para buscar e armazenar terceirizados no PostgreSQL do MIR."""
+
+    @task
+    def fetch_terceirizados() -> None:
+        logging.info("[contratos_terceirizados_mir_ingest_dag.py] Starting fetch_terceirizados task")
+        api = ClienteContratos()
+        postgres_conn_str = get_postgres_conn("postgres_mir")  
+        db = ClientPostgresDB(postgres_conn_str)
+        contratos_ids = db.get_contratos_ids()
+
+        for contrato_id in contratos_ids:
+            try:
+                logging.info(
+                    f"[contratos_terceirizados_mir_ingest_dag.py] Fetching terceirizados for contrato ID: "
+                    f"{contrato_id}"
+                )
+                terceirizados = api.get_terceirizados_by_contrato_id(str(contrato_id))
+
+                if terceirizados:
+                    for fatura in terceirizados:
+                        fatura["dt_ingest"] = datetime.now().isoformat()
+
+                db.insert_data(
+                    terceirizados,
+                    "terceirizados",
+                    conflict_fields=["id"],
+                    primary_key=["id"],
+                    schema="compras_gov",
+                )
+            except Exception as e:
+                logging.error(
+                    f"[contratos_terceirizados_mir_ingest_dag.py] Error for contrato ID {contrato_id}: {e}"
+                )
+
+    fetch_terceirizados()
+
+
+dag_instance = api_terceirizados_mir_dag()