feat: pipeline orientado a eventos (check_update -> download -> mat_test)

.github/scripts/deploy_flows.py

@@ -193,6 +193,9 @@ def deploy_flow(
 
     job_variables = getattr(flow, "job_variables", None)
 
+    extra_tags = getattr(flow, "deploy_tags", None) or []
+    tags = ["automated-deploy", *extra_tags]
+
     try:
         flow.from_source(
             source=GitRepository(
@@ -203,7 +206,7 @@ def deploy_flow(
         ).deploy(
             name=deployment_name,
             work_pool_name=pool_name,
-            tags=["automated-deploy"],
+            tags=tags,
             schedules=schedules,
             job_variables=job_variables,
             build=False,
@@ -214,7 +217,7 @@ def deploy_flow(
             if not schedules
             else f"com schedules: {schedules}"
         )
-        return True, f"  ✓ {deployment_name} registrado {status}"
+        return True, f"  ✓ {deployment_name} registrado {status}, tags={tags}"
     except Exception as e:
         return False, f"  ✗ Falha ao registrar {deployment_name}: {e}"
 

models/test_dataset/test_dataset__test_event_pipeline.sql

@@ -0,0 +1,13 @@
+{{
+    config(
+        schema="test_dataset",
+        alias="test_event_pipeline",
+        materialized="table",
+    )
+}}
+
+-- Piloto da issue #1867: lê o CSV que flow_download_flow sobe pro staging
+-- (upload_to_gcs) — cada rodada substitui a única linha pela data do
+-- download mais recente.
+select safe_cast(reference_date as date) as reference_date
+from {{ set_datalake_project("test_dataset_staging.test_event_pipeline") }} as t

models/test_dataset/test_dataset__test_event_pipeline_partitioned.sql

@@ -0,0 +1,20 @@
+{{
+    config(
+        schema="test_dataset",
+        alias="test_event_pipeline_partitioned",
+        materialized="table",
+    )
+}}
+
+-- Variante particionada (ano=/mes=) do piloto da issue #1867 — testa
+-- DownloadResult.partition_folders/transfer_files_to_prod_flow(folders=...)
+-- promovendo só a fatia nova, não o staging inteiro. As colunas ano/mes
+-- vêm explícitas no CSV (não extraídas do caminho Hive), igual ao dado
+-- real de cada arquivo.
+select
+    safe_cast(ano as int64) as ano,
+    safe_cast(mes as int64) as mes,
+    safe_cast(reference_date as date) as reference_date
+from
+    {{ set_datalake_project("test_dataset_staging.test_event_pipeline_partitioned") }}
+    as t

pipelines/datasets/br_ans_beneficiario/constants.py

@@ -0,0 +1,8 @@
+"""
+Constant values for br_ans_beneficiario.
+"""
+
+DATASET_ID = "br_ans_beneficiario"
+
+INFORMACAO_CONSOLIDADA_TABLE_ID = "informacao_consolidada"
+INFORMACAO_CONSOLIDADA_URL = "https://dadosabertos.ans.gov.br/FTP/PDA/informacoes_consolidadas_de_beneficiarios-024/"

pipelines/datasets/br_ans_beneficiario/flows.py

@@ -1,150 +1,74 @@
 """
-Flow br_ans_beneficiario__informacao_consolidada — Prefect 3.
+Flows para br_ans_beneficiario — Prefect 3.
+
+Migrado por completo pro pipeline orientado a eventos (issue #1867):
+check_update -> download -> mat_test. Lógica específica do dataset mora em
+`tasks.py`, constantes em `constants.py` — aqui só a fiação
+(`CheckThenDownloadPipeline` + `@flow`).
 """
 
 from prefect import flow
 
-from pipelines.crawler.ans_beneficiario.tasks import (
-    crawler_ans,
-    extract_links_and_dates,
-    files_to_download,
-    get_file_max_date,
+from pipelines.datasets.br_ans_beneficiario.constants import (
+    DATASET_ID,
+    INFORMACAO_CONSOLIDADA_TABLE_ID,
 )
-from pipelines.utils.metadata.domain import (
-    DateFormat,
-    PartBdpro,
-    YearMonth,
+from pipelines.datasets.br_ans_beneficiario.tasks import (
+    br_ans_beneficiario_check_for_update,
+    br_ans_beneficiario_download,
 )
-from pipelines.utils.metadata.tasks import (
-    commit_source_update_task,
-    poll_source_for_update_task,
-    register_table_materialization_task,
+from pipelines.utils.stage_dispatch import (
+    CheckThenDownloadPipeline,
+    Etapa,
+    deploy_tags,
 )
-from pipelines.utils.tasks import (
-    rename_flow_run_dataset_table,
-    run_dbt,
-    upload_to_gcs,
+
+_informacao_consolidada_pipeline = CheckThenDownloadPipeline(
+    dataset_id=DATASET_ID,
+    table_id=INFORMACAO_CONSOLIDADA_TABLE_ID,
+    check_for_update=br_ans_beneficiario_check_for_update,
+    download_data=br_ans_beneficiario_download,
+    # Mesma granularidade do flow antigo: dado é mensal, sem dia.
+    date_format="%Y-%m",
 )
 
 
 @flow(
-    name="br_ans_beneficiario__informacao_consolidada",
+    name=_informacao_consolidada_pipeline.check_update_flow_name,
     log_prints=True,
 )
-def br_ans_beneficiario__informacao_consolidada(
-    dataset_id: str = "br_ans_beneficiario",
-    table_id: str = "informacao_consolidada",
-    url: str = "https://dadosabertos.ans.gov.br/FTP/PDA/informacoes_consolidadas_de_beneficiarios-024/",
-    year: str | None = None,
-    materialize_after_dump: bool = True,
-    update_metadata: bool = True,
-    target: str = "prod",
-    force_run: bool = False,
-) -> None:
-    # pyrefly: ignore [unused-coroutine]
-    rename_flow_run_dataset_table(
-        prefix="Dump: ", dataset_id=dataset_id, table_id=table_id
-    )
-
-    links_and_dates = extract_links_and_dates(url=url)
-    file_last_date = get_file_max_date(df=links_and_dates)
-
-    if force_run:
-        files = files_to_download(df=links_and_dates, year=year)
-    else:
-        has_new_data = poll_source_for_update_task(
-            dataset_id=dataset_id,
-            table_id=table_id,
-            source_max_date=file_last_date,
-            env="prod",
-            date_format="%Y-%m",
-            compare_against="coverage",
-        )
-        if not has_new_data:
-            print(f"Não há atualizações para a tabela {table_id}!")
-            return
-        files = files_to_download(df=links_and_dates, year=None)
-
-    # Comita o Update da fonte já aqui, antes de baixar/materializar: se o
-    # flow falhar no meio, o metadado da fonte ainda reflete que havia dado
-    # novo publicado, mesmo que a tabela não tenha sido atualizada.
-    commit_source_update_task(
-        dataset_id=dataset_id,
-        table_id=table_id,
-        source_max_date=file_last_date,
-        env="prod",
-        date_format="%Y-%m",
-        update_metadata=update_metadata,
-        materialize_after_dump=materialize_after_dump,
-    )
-
-    if not files:
-        print("Nenhum arquivo para baixar.")
-        return
-
-    output_filepath = crawler_ans(files=files)
-
-    # crawler_ans -> parquet_partition grava .parquet. Sem declarar o
-    # formato, o dump_header chamado por upload_to_gcs procura .csv (default)
-    # e não encontra nada.
-    upload_to_gcs(
-        data_path=output_filepath,
-        dataset_id=dataset_id,
-        table_id=table_id,
-        bucket_name="basedosdados-dev",
-        dump_mode="append",
-        source_format="parquet",
-    )
+def br_ans_beneficiario_informacao_consolidada_check_update() -> None:
+    _informacao_consolidada_pipeline.run_check_update()
 
-    run_dbt(
-        dataset_id=dataset_id,
-        table_id=table_id,
-        dbt_command="run/test",
-        target="dev",
-    )
 
-    if not materialize_after_dump:
-        return
-
-    upload_to_gcs(
-        data_path=output_filepath,
-        dataset_id=dataset_id,
-        table_id=table_id,
-        bucket_name="basedosdados",
-        dump_mode="append",
-        source_format="parquet",
-    )
+# pyrefly: ignore [missing-attribute]
+br_ans_beneficiario_informacao_consolidada_check_update.deploy_tags = (
+    deploy_tags(DATASET_ID, Etapa.CHECK_UPDATE)
+)
 
-    run_dbt(
-        dataset_id=dataset_id,
-        table_id=table_id,
-        dbt_command="run/test",
-        target=target,
-    )
 
-    if update_metadata:
-        # R4-normalização: o legado usava date_format="%Y-%m-%d" com colunas
-        # ano/mes (emitia um endDay/startDay=1 espúrio). O domínio refatorado
-        # força YearMonth↔YEAR_MONTH, descartando o dia — granularidade correta
-        # para dado mensal. Valores ano/mês idênticos ao legado; free_lag default
-        # (months=6) reproduz time_delta={"months":6}.
-        register_table_materialization_task(
-            dataset_id=dataset_id,
-            table_id=table_id,
-            coverage=PartBdpro(
-                date_column=YearMonth(year="ano", month="mes"),
-                date_format=DateFormat.YEAR_MONTH,
-            ),
-            env="prod",
-            bq_project="basedosdados",
-        )
+@flow(
+    name=_informacao_consolidada_pipeline.download_flow_name,
+    log_prints=True,
+)
+def br_ans_beneficiario_informacao_consolidada_download(
+    download_params: dict,
+) -> None:
+    _informacao_consolidada_pipeline.run_download(download_params)
 
 
 # pyrefly: ignore [missing-attribute]
-br_ans_beneficiario__informacao_consolidada.deploy_schedules = [
-    {"cron": "0 21 * * *", "timezone": "America/Sao_Paulo"}
-]
+br_ans_beneficiario_informacao_consolidada_download.deploy_tags = deploy_tags(
+    DATASET_ID, Etapa.DOWNLOAD
+)
 # Pico medido em produção após otimizar parquet_partition (category dtype +
-# del/gc.collect() por estado): ~1.78Gi. ~1.7x de margem sobre esse valor.
+# del/gc.collect() por estado): ~1.78Gi. ~1.7x de margem sobre esse valor —
+# mesmo tier do flow antigo, já que o download pesado (crawler_ans) continua
+# acontecendo aqui.
 # pyrefly: ignore [missing-attribute]
-br_ans_beneficiario__informacao_consolidada.job_variables = {"memory": "3Gi"}
+br_ans_beneficiario_informacao_consolidada_download.job_variables = {
+    "memory": "3Gi"
+}
+_informacao_consolidada_pipeline.download_deployment = (
+    br_ans_beneficiario_informacao_consolidada_download.fn.__name__
+)

pipelines/datasets/br_ans_beneficiario/tasks.py

@@ -0,0 +1,60 @@
+"""
+Tasks for br_ans_beneficiario.
+"""
+
+from datetime import date
+
+from pipelines.crawler.ans_beneficiario.tasks import (
+    crawler_ans,
+    extract_links_and_dates,
+    files_to_download,
+    get_file_max_date,
+)
+from pipelines.datasets.br_ans_beneficiario.constants import (
+    INFORMACAO_CONSOLIDADA_URL,
+)
+from pipelines.utils.metadata.domain import DateFormat, PartBdpro, YearMonth
+from pipelines.utils.stage_dispatch import CheckResult, DownloadResult
+
+# ──────────────────────────────────────────────────────────────────────────────
+# informacao_consolidada (issue #1867) — ver constants.py
+#
+# Diferente de br_ibge_ipca: aqui o check É leve de verdade e independente —
+# `extract_links_and_dates` só faz um GET na página de listagem (HTML da
+# pasta FTP-like da ANS) e lê datas de "última atualização" já presentes no
+# próprio HTML, sem baixar nenhum arquivo de dado. `download_data` refaz essa
+# mesma listagem (idem barato) só pra descobrir de novo quais arquivos
+# baixar — não dá pra repassar o DataFrame entre pods, mas a chamada em si
+# é a mesma leve de sempre, não o download pesado (que só acontece depois,
+# em `crawler_ans`).
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def br_ans_beneficiario_check_for_update() -> CheckResult:
+    links_and_dates = extract_links_and_dates(url=INFORMACAO_CONSOLIDADA_URL)
+    file_last_date = get_file_max_date(df=links_and_dates)  # "YYYY-MM"
+    reference_date = date.fromisoformat(f"{file_last_date}-01")
+    return CheckResult(reference_date=reference_date)
+
+
+def br_ans_beneficiario_download(download_params: dict) -> DownloadResult:
+    links_and_dates = extract_links_and_dates(url=INFORMACAO_CONSOLIDADA_URL)
+    files = files_to_download(df=links_and_dates, year=None)
+
+    output_filepath = crawler_ans(files=files)
+
+    ref = date.fromisoformat(download_params["reference_date"])
+    return DownloadResult(
+        coverage=PartBdpro(
+            date_column=YearMonth(year="ano", month="mes"),
+            date_format=DateFormat.YEAR_MONTH,
+        ).model_dump(),
+        data_path=output_filepath,
+        bq_project="basedosdados",
+        # crawler_ans -> parquet_partition grava .parquet, não .csv (default).
+        source_format="parquet",
+        # to_partitions particiona por ano/mes/sigla_uf/modalidade_operadora,
+        # mas só um ano/mes muda por execução — promover o nível ano/mes já
+        # carrega todas as sub-partições de uf/modalidade daquele mês.
+        partition_folders=[f"ano={ref.year}/mes={ref.month:02d}"],
+    )

pipelines/datasets/br_ibge_ipca/constants.py

@@ -0,0 +1,14 @@
+"""
+Constant values for br_ibge_ipca.
+"""
+
+DATASET_ID = "br_ibge_ipca"
+
+# As 4 tabelas do dataset — migradas pro pipeline orientado a eventos
+# (issue #1867), substituindo o antigo `_ipca_flow`/`_run_ibge_inflacao`
+# monolítico (que segue existindo em `crawler/ibge_inflacao/flows.py`,
+# ainda usado por br_ibge_ipca15/br_ibge_inpc).
+MES_BRASIL_TABLE_ID = "mes_brasil"
+MES_CATEGORIA_BRASIL_TABLE_ID = "mes_categoria_brasil"
+MES_CATEGORIA_RM_TABLE_ID = "mes_categoria_rm"
+MES_CATEGORIA_MUNICIPIO_TABLE_ID = "mes_categoria_municipio"