feat: pipeline orientado a eventos (check_update -> flow_download -> mat_test) by Winzen · Pull Request #1932 · basedosdados/pipelines

Winzen · 2026-08-30T23:39:39Z

[Feature]: piloto de infraestrutura pra provar a mecânica da [chore] Pipeline orientado a eventos #1867 de ponta a ponta, usando datasets semi-sintéticos (test_dataset.test_event_pipeline e test_dataset.test_event_pipeline_partitioned, cadastro real no backend/prod, dado real no BigQuery de dev e de prod).

Relacionado à #1867.

Descrição do PR

Piloto completo (não mais só scaffold) do que a #1867 propõe: quebrar um flow monolítico em 3 flows independentes — check_update → flow_download → mat_test (genérico, compartilhado por qualquer dataset) — testado de ponta a ponta com infraestrutura real em dev e em produção, não simulado.

Mudança em relação ao design original desta PR: as etapas não são mais encadeadas por Automação do Prefect 3 (emit_event + EventTrigger + RunDeployment) — são disparadas por run_deployment() chamado direto no código do flow upstream. Motivo e comparação completa: pipelines/utils/stage_dispatch.py e a documentação linkada abaixo. Resumo: mesmo isolamento de recurso entre pods (nenhum bloqueio no disparo), mas sem precisar manter um mecanismo separado de automações sincronizadas por dataset, e com validação de parâmetro nativa do Pydantic (sem o workaround de serializar tudo em JSON pra contornar o Jinja da automação).

Desde a versão inicial deste PR, o boilerplate compartilhado foi encapsulado numa classe genérica (CheckThenDownloadPipeline) e testado também com dados particionados, além de reorganizado pra seguir a convenção real de pasta-por-dataset_id do repo — ver "Detalhes Técnicos" e "Teste e Validações" abaixo.

Detalhes Técnicos

pipelines/utils/stage_dispatch.py:
- CheckThenDownloadPipeline — encapsula o boilerplate compartilhado entre check_update/flow_download (rename do flow run, poll/commit de coverage, dispatch pro próximo estágio via run_deployment()) numa classe reutilizável. Cada dataset novo só fornece check_fn/download_fn com a lógica específica.
- CheckResult/DownloadResult — contratos tipados de dados entre os estágios check/download.
- Etapa(StrEnum) — substitui strings soltas (check_update/flow_download/mat_test) repetidas em deployment_name/deploy_tags, evitando typo silencioso que só falhava tarde, no run_deployment().
- deployment_name()/CheckThenDownloadPipeline ganharam um parâmetro opcional flow_download_deployment, setado como atributo a partir do .fn.__name__ do flow real (não uma string repetida solta) — necessário pra suportar múltiplos pilotos/tabelas convivendo no mesmo flows.py (ver próximo bullet), já que deploy_flows.py usa o nome da variável do módulo como nome literal do deployment.
- deploy_tags simplificado: a tag da etapa agora é só o nome dela (check_update), sem o prefixo etapa: que só poluía.
pipelines/datasets/test_dataset/ (constants.py/tasks.py/flows.py únicos, seções comentadas por piloto — segue a convenção real do repo de pasta-por-dataset_id, ver br_ms_sih/flows.py):
- event_pipeline — piloto original: check_update_flow (checagem real via poll_source_for_update_task/commit_source_update_task, mesmo padrão dos datasets reais) dispara flow_download via run_deployment() só se houver dado novo; flow_download_flow cria e sobe um CSV real pro staging (upload_to_gcs), dispara mat_test.
- event_pipeline_partitioned — variante nova testando dados particionados (ano=/mes=, 2 níveis Hive), exercitando DownloadResult.partition_folders/transfer_files_to_prod_flow(folders=...) pela primeira vez (o piloto original usa um único arquivo sem partição, nunca cobriu esse caminho).
pipelines/utils/metadata/flows.py::mat_test_flow — genérico, um deployment só, reaproveitado por qualquer dataset: dbt run+test em dev → promove pra prod (transfer_files_to_prod_flow, subflow já existente no repo, nunca chamado antes desta issue) → dbt run+test em prod → atualiza coverage no backend (register_table_materialization_task, chamado direto pra manter o retry da task).
4 problemas reais achados e corrigidos testando o piloto particionado, 2 deles genéricos e não específicos deste piloto:
- poll_source_for_update quebra pra qualquer tabela sem RawDataSource vinculado — o filtro GraphQL rawDataSource_Id: null é ignorado (não filtra por nulo), retornando todos os Polls do backend e estourando a checagem de "no máximo 1 resultado". Ainda sem issue própria.
- bd.Table.create() só registra o ponteiro externo do BigQuery em basedosdados-staging, nunca em basedosdados-dev — o target=dev do dbt espera achar a tabela lá, exigindo criação manual não documentada até então (o piloto original já dependia disso, sem estar no código) — issue bd.Table.create() só registra o ponteiro externo do BigQuery em basedosdados-staging, nunca em basedosdados-dev #1967.
- Erro próprio: ano/mes duplicados dentro do CSV, violando a convenção Hive (a coluna de partição só deveria existir no caminho).
- Tabela cadastrada como published em vez de under_review, travando assert_write_allowed.
Mudanças nos dados e no schema: tabelas reais criadas e materializadas em basedosdados-dev/basedosdados (produção real), test_dataset.test_event_pipeline e test_dataset.test_event_pipeline_partitioned — dado de teste (reference_date, ano, mes), sem impacto em datasets reais.
Impacto no desempenho: nenhum em produção real — só os datasets de teste deste piloto materializam.

⚠️ Este PR mantém deployments reais registrados no pool basedosdados-dev (sem schedule) e basedosdados (mat_test genérico, sem schedule, só acionado por run_deployment()).

Teste e Validações

Testado localmente
Testado na Cloud (dev)
Testado na Cloud (prod)

Cadeia completa (check_update → flow_download → mat_test) validada de ponta a ponta múltiplas vezes, incluindo:

Detecção real de dado desatualizado (coverage do backend comparada contra a data de hoje) e o caminho "sem atualização" (nada dispara).
Promoção real dev → prod (transfer_files_to_prod_flow), com as tabelas basedosdados.test_dataset.test_event_pipeline e basedosdados.test_dataset.test_event_pipeline_partitioned materializadas de verdade em produção, usando a service account real (dbt-rpc@basedosdados.iam.gserviceaccount.com).
Troca do mecanismo de disparo (Automação → run_deployment()) retestada de ponta a ponta depois de implementada — sucesso na primeira tentativa, incluindo lineage real confirmado no Prefect UI ("Beginning subflow run").
Variante particionada testada de ponta a ponta (ano=/mes=), incluindo o caminho transfer_files_to_prod_flow(folders=...) que promove só a fatia nova, não o staging inteiro — nunca exercitado antes deste PR.
Depois da consolidação em test_dataset/ (dois pilotos no mesmo flows.py), reteste forçando dado novo de propósito (coverage rolada pra ontem no backend) pra validar que o dispatch automático resolve o flow_download certo pelo nome derivado (flow_download_deployment/.fn.__name__) e não pelo nome-padrão adivinhado: check_update (ambos os pilotos, detectou has_new_data=True) → flow_download (ambos, nome novo) → mat_test (ambos) — todos COMPLETED.

Documentação completa (log cronológico, comparações arquiteturais, fluxogramas) em D:\repositorios\ftwca\tasks\pipelines\issue_1867\ (repo pessoal de documentação, não faz parte deste PR).

Riscos e Mitigações

Riscos conhecidos: os datasets de teste (test_dataset.test_event_pipeline, test_dataset.test_event_pipeline_partitioned) têm tabelas reais materializadas em produção real (basedosdados.test_dataset.*), criadas durante a validação — é dado de teste, mas existe de fato em prod. Cleanup ainda não decidido (ver issue [chore] Pipeline orientado a eventos #1867, "Novidades não previstas").
Planos de rollback: deletar as tabelas de teste em prod/dev, os deployments registrados (test_event_pipeline*: *, mat_test/mat_test_flow, update_temporal_coverage), e o registro do dataset/tabelas no backend.

Dependências

Nenhuma dependência bloqueante. Issues de acompanhamento abertas a partir deste trabalho, não bloqueantes pra este PR: rename_flow_run_dataset_table nunca renomeia o flow run (async task chamada sem await) #1940 (bug repo-wide do rename), Deploy de prod (cd-prefect3.yaml) deveria ser seletivo, não --all em todo push #1943 (deploy de prod seletivo, necessário antes de escalar pros ~82 datasets reais), bd.Table.create() só registra o ponteiro externo do BigQuery em basedosdados-staging, nunca em basedosdados-dev #1967 (ponteiro do BigQuery ausente em basedosdados-dev). O bug do allPoll(rawDataSource_Id: null) (ver "Detalhes Técnicos") ainda não virou issue separada.

…mations (#1867) Adiciona check_update_flow e flow_download_flow sintéticos pra provar, ponta a ponta, a mecânica de automações do Prefect 3 proposta na #1867: evento custom (emit_event) disparando automação que roda o flow seguinte com parâmetros computados em runtime, propagados via Jinja no payload. Sem dataset real, sem tocar BigQuery/GCS/dbt.

coderabbitai · 2026-08-30T23:39:50Z

Important

Draft PR not reviewed

Draft PRs are not automatically reviewed by default.

Trigger a manual review

To automatically review draft PRs, update your CodeRabbit configuration:

reviews:
  auto_review:
    drafts: true

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

_{Comment @coderabbitai help to get the list of available commands.}

…ence_date, add deploy_tags - check_update_flow agora calcula reference_date em runtime (não repassa mais como parâmetro de entrada), e monta um dict download_params serializado em JSON, não campos fixos — automação passa a referenciar só esse campo via Jinja. - Novo módulo pipelines/utils/automations.py: convenções compartilhadas (dataset_resource_id, event_name, encode_params/decode_params, deploy_tags, build_chained_automation) pra automações não duplicarem strings. - deploy_flows.py passa a ler flow.deploy_tags (aditivo). - scripts/pilot_event_automations.py: cria ou atualiza a Automação 1.

for more information, see https://pre-commit.ci

…wnload -> mat_test) - mat_test_flow: etapa terminal, recebe mat_test_params (JSON) via Automação 2, decodifica e loga — não emite evento (fim da cadeia). - flow_download_flow agora emite flow_download.completed ao final, repassando reference_date + download_path pro mat_test via emit_flow_download_completed. - scripts/pilot_event_automations.py generalizado pra criar/atualizar as duas automações da cadeia numa lista, não uma função hardcoded por automação.

for more information, see https://pre-commit.ci

…0 compat pre-commit.ci's ruff auto-fix rewrote timezone.utc as datetime.UTC, which only exists in Python 3.11+. This repo's pyproject.toml declares requires-python ">=3.10,<3.13" but ruff's target-version is "py312", so the autofix broke deploy_flows.py's import on this 3.10 environment. Reverting locally; the target-version mismatch in pyproject.toml is a separate, repo-wide issue not addressed here.

for more information, see https://pre-commit.ci

- Posture.Reactive (enum) instead of the string "Reactive" in build_chained_automation's EventTrigger. - # pyrefly: ignore [missing-attribute] on the three <flow>.deploy_tags assignments (same convention already used for deploy_schedules elsewhere) and on automation.create()'s .name/.id access (Automation.create is @async_dispatch-decorated, which pyrefly can't resolve past). Verified locally with `uv run pyrefly check`: all 6 errors from CI run 33447481821 are gone; the 7 remaining diagnostics are pre-existing on main, in files this PR doesn't touch.

… name build_chained_automation now requires the triggering event's related resources to include prefect.tag.etapa:<upstream_etapa> (attached automatically by Prefect from the deployment's deploy_tags), in addition to the existing match on resource id + event name. Makes each automation provably scoped to its own etapa, not just relying on the event name string to disambiguate. Verified: re-ran the full 3-hop chain (check_update -> flow_download -> mat_test) — both automations still fire exactly once each, no cross-firing, confirmed via prefect.automation.triggered events.

… + generic mat_test_flow - check_update_flow: real poll against the backend Coverage for test_dataset.test_event_pipeline (poll_source_for_update_task), not a simulated bool. All poll+commit+emit logic centralized in a new check_update_and_emit helper (pipelines/utils/automations.py), reusable by any dataset's check_update. - flow_download_flow: only downloads (writes a small CSV) and updates staging (upload_to_gcs) — no materialization here anymore. - New generic mat_test_flow (pipelines/utils/metadata/flows.py): runs dbt run/test for dev and prod, then updates the backend coverage via the existing update_temporal_coverage flow (as a subflow) — reused by any dataset's Automation 2, not one mat_test_flow per dataset. All per-dataset specifics (dataset_id, table_id, coverage spec, env, bq_project, prefect_mode) travel via the event payload. update_temporal_coverage gained a prefect_mode param (additive). - New dbt model models/test_dataset/test_dataset__test_event_pipeline.sql, reading from staging (uploaded by flow_download_flow). - scripts/pilot_event_automations.py: Automation 2 now targets the shared mat_test/mat_test_flow deployment instead of a per-dataset one. Backend: registered test_dataset.test_event_pipeline (status under_review, required for prod-metadata writes against a non-prod bq_project) with a RawDataSource, a column, and a Coverage deliberately ending yesterday, to prove the real "outdated" detection on first run.

for more information, see https://pre-commit.ci

…promotion - mat_test_flow always runs run_dbt(target="dev") first; a failure there raises and aborts the flow before anything touches prod (same guarantee as `if not materialize_after_dump` in the single-flow real pattern). Only when "prod" is in targets does it call transfer_files_to_prod_flow (already existing in pipelines/utils/materialize_prod/) as a subflow — reused, not reimplemented — to download the already-uploaded dev staging data and push it into the real prod bucket/project, then run_dbt(target="prod"). - transfer_files_to_prod_flow / download_files_from_bucket_folders: added support for tables with no Hive-style partition (folders=None downloads directly from the table's staging prefix, no subfolder) — needed for tables like the pilot's test_event_pipeline. Also removed a hardcoded placeholder ("mes_competencia=202306...") that silently overrode a caller's folders=None, since this flow is now called programmatically, not just triggered manually from the UI. - mat_test_flow reads an optional `partition_folders` from the event payload and passes it straight through, so a partitioned dataset only promotes the slice flow_download just updated, not the whole staging folder. Verified: read-only IAM check confirms the dev service account (chave-subidores-de-dados@basedosdados-dev) only has storage.objects.get/ list on the real `basedosdados` bucket, no write — so this path cannot be exercised end-to-end without real prod credentials. test_event_pipeline still runs with targets=["dev"] only, so this code path isn't hit by the pilot yet; documented in D:\docs\pipelines\staging-multi-ambiente.md.

transfer_files_to_prod_flow was hardcoded to dbt_command="run" for the prod materialization — meaning it never actually ran dbt tests against prod. Turned into a parameter (default "run", preserving current behavior for existing/manual callers) so mat_test_flow can pass "run/test" explicitly, since testing before publishing is the whole point of that flow.

for more information, see https://pre-commit.ci

mat_test_flow is about to be redeployed to the basedosdados (prod) work pool so it can actually reach real prod GCS/BigQuery for the dev->prod promotion. The requester-pays billing project for reading the dev bucket needs to be a project the running pod's own service account has serviceusage.services.use on — that's now basedosdados (the pool the pod runs in), not basedosdados-dev.

TARGETS -> ["dev", "prod"] now that mat_test_flow's deployment runs from the basedosdados (prod) work pool, which should have real write access to the basedosdados project. First real test of transfer_files_to_prod_flow end-to-end.

mat_test_flow now takes dataset_id/table_id directly instead of only inside the mat_test_params JSON blob — they're plain strings, so unlike coverage (a nested object) there's no Jinja type-loss risk in passing them as their own automation parameters. Gets the flow run renamed via rename_flow_run_dataset_table (same pattern used elsewhere in the repo), so runs are identifiable in the Prefect UI without opening logs. Also: mat_test_flow now calls register_table_materialization_task directly instead of going through the update_temporal_coverage flow wrapper, which has no retry configured on its own @flow — going through it silently dropped the @task-level retries/retry_delay that register_table_materialization_task already has. update_temporal_coverage itself is untouched and still usable standalone.

for more information, see https://pre-commit.ci

@task

…nchronously register_table_materialization_task now called directly (not through the update_temporal_coverage flow, which no longer runs), so it stopped getting Pydantic's automatic parameter coercion — the JSON-decoded coverage dict never became a real CoverageSpec instance, failing at coverage.date_format. rename_flow_run_dataset_table is an async @task; calling it unawaited from a sync flow just creates a coroutine that's silently discarded (confirmed empirically against real production flow run names, e.g. br_bcb_estban) — wrap it in run_coro_as_sync instead.

@task

Same latent bug as mat_test_flow: calling an async @task unawaited from a sync flow just creates a coroutine that's silently discarded, so the flow run never actually got renamed — it kept its auto-generated name instead of "Materialização Prod: <dataset_id>.<table_id>".

Removed the stale comparison against update_temporal_coverage (no longer called by this flow) and documented the two behaviors added since the docstring was last written: run_coro_as_sync around the rename call, and the TypeAdapter coercion of the coverage dict before register_table_materialization_task.

…nt() Investigated in run-deployment-vs-automacao.md: since run_deployment()'s timeout=0 doesn't block the caller (same resource-isolation properties as the Automation-triggered RunDeployment action), the 3-deployment topology stays the same, but dispatch now happens as a direct call inside the upstream flow instead of a separately-maintained Automation object. This avoids having to build a whole tag-discovery-and-sync mechanism to scale automations to the ~82 real datasets (automacoes-em-massa.md becomes moot), and restores native parameter typing (mat_test_flow's coverage is CoverageSpec again, validated by Pydantic automatically — no more manual TypeAdapter workaround, no more JSON-string-in-a-string-param via encode_params/decode_params, since run_deployment()'s parameters dict isn't limited by Automations' Jinja string-only rendering). - pipelines/utils/automations.py: check_update_and_emit -> check_update_and_dispatch (calls run_deployment(timeout=0, as_subflow=True) instead of emit_event). Dropped Automation/EventTrigger/ResourceSpecification/emit_event/ encode_params/decode_params/payload_parameters/build_chained_automation. Added deployment_name() to resolve "<flow name>/<deployment name>" by convention. Kept deploy_tags()/etapa_tag() as an organizational convention (no longer functionally required). - pipelines/datasets/test_event_pipeline/tasks.py: emit_flow_download_completed -> dispatch_mat_test, calling run_deployment() with typed kwargs (coverage as dict, deserialized by mat_test_flow's typed parameter). - pipelines/datasets/test_event_pipeline/flows.py: flow_download_flow's download_params is now dict, not a JSON string. - pipelines/utils/metadata/flows.py: mat_test_flow's mat_test_params (JSON string) decomposed into typed kwargs (coverage: CoverageSpec, env, bq_project, prefect_mode, targets, partition_folders, download_billing_project) — drops decode_params and the TypeAdapter workaround entirely. - scripts/pilot_event_automations.py: removed, no longer needed.

Follow-up to the run_deployment() switch (5c6412a) — the module and a few names/comments still carried vocabulary from the removed Automation/emit_event mechanism. - pipelines/utils/automations.py -> pipelines/utils/stage_dispatch.py: the module isn't about Automations anymore, just dispatch between pipeline stages. - check_update_and_dispatch: dropped upstream_etapa (dead — was only used to build the emit_event event name, never read in the current body). Renamed resource_dataset_id -> dataset_id and downstream_etapa -> next_etapa (no longer describes an event-resource relationship, just "which deployment to call next"). - Updated all call sites and stale docstring/comment references (constants.py's "resource id" comment, file-path pointers to the old automations.py, historical "isso só era necessário quando... Jinja das automações" justification trimmed to state the current behavior).

Requested cleanup: dataset_id/table_id should mean the real backend/BQ identity everywhere, matching the rest of the repo's convention (register_table_materialization_task, run_dbt, upload_to_gcs all already use plain dataset_id/table_id for this). The "backend_" prefix only existed to disambiguate from the Prefect deployment-naming convention sharing the same name — renamed that one to prefect_dataset_id/ PREFECT_DATASET_ID instead, since it's the one with the narrower, Prefect-specific meaning. - constants.py: BACKEND_DATASET_ID/BACKEND_TABLE_ID -> DATASET_ID/TABLE_ID; DATASET_ID (deployment-naming) -> PREFECT_DATASET_ID. - stage_dispatch.py: check_update_and_dispatch's dataset_id (deployment naming) -> prefect_dataset_id; backend_dataset_id/backend_table_id -> dataset_id/table_id. - tasks.py: dispatch_mat_test's backend_dataset_id/backend_table_id -> dataset_id/table_id. - flows.py: check_update_flow and flow_download_flow now also call rename_flow_run_dataset_table (via run_coro_as_sync, issue #1940) — they never had it before, unlike mat_test_flow/transfer_files_to_prod_flow.

for more information, see https://pre-commit.ci

…loadPipeline Introduz CheckThenDownloadPipeline (pipelines/utils/stage_dispatch.py), a interface recomendada pra datasets na variante padrão (check_update e flow_download separados): encapsula o boilerplate repetido entre os dois estagios (rename do flow run, poll/commit/dispatch, dispatch pro mat_test) atras de dois contratos tipados, CheckResult e DownloadResult. Cada dataset so fornece check_fn/download_fn com a logica especifica dele; o @flow em si continua fino e no proprio flows.py do dataset (deploy_flows.py so reconhece Flow cuja funcao esta definida no arquivo do dataset). test_event_pipeline (piloto da issue #1867) refatorado pra usar a nova interface, sem mudanca de comportamento. Adiciona job_variables por etapa (issue #1867: recursos identicos pra cargas diferentes era um dos problemas motivadores). check_update e flow_download variam por dataset e ficam declarados no constants.py de cada um; mat_test e um deployment unico compartilhado, entao o tier dele mora em pipelines/utils/metadata/constants.py. mat_test em 2Gi e provisorio, baseado em uso real medido (~1GB) no teste ponta a ponta de 2026-09-01 — reavaliar quando um dataset real (nao sintetico) passar por aqui, especialmente em source_format=parquet (dump_header le um row group inteiro pra inferir schema nesse formato). Adiciona Etapa(StrEnum) pras tres etapas (check_update/flow_download/ mat_test), substituindo string solta em deployment_name/deploy_tags/ etapa_tag — evita que um typo silencie um nome de deployment errado em vez de falhar na hora.

…prod test_event_pipeline_partitioned: variante do piloto original testando DownloadResult.partition_folders e transfer_files_to_prod_flow(folders=...) de ponta a ponta — o piloto original (test_event_pipeline) usa um unico arquivo sem particao, entao nunca exercitou esse caminho. Reaproveita CheckThenDownloadPipeline/Etapa/mat_test_flow genericos, mesmo dataset test_dataset no backend (tabela nova). Registrado de verdade em prod: tabela test_event_pipeline_partitioned com colunas ano/mes marcadas is_partition, coverage, cloud table, update record.

write_partitioned_csv escrevia ano/mes como colunas do arquivo, além de codificar no caminho da pasta (ano=/mes=). Convenção Hive é a coluna de partição existir só no caminho -- repetir no arquivo colide com o que o particionamento automático do BigQuery (bd.Table.create(), mode=STRINGS) já adiciona, e o BigQuery descarta a duplicata do schema declarado, deixando só reference_date como coluna de arquivo esperada: com 2 valores a mais por linha, "Database Error: Too many values in line. Found 3 column(s) when expecting 1."

Move test_event_pipeline/ e test_event_pipeline_partitioned/ pra dentro de pipelines/datasets/test_dataset/ (constants.py/tasks.py/flows.py compartilhados, secoes com banner de comentario por piloto) -- as duas tabelas pertencem ao mesmo dataset_id (test_dataset) no backend, entao duas pastas separadas fugia da convencao real do repo (ver pipelines/datasets/br_ms_sih/flows.py: varias tabelas, um flows.py so). Generaliza deployment_name()/CheckThenDownloadPipeline pra suportar isso: antes o nome do deployment assumia que a variavel do flow se chamava literalmente <etapa>_flow, o que so funciona com um dataset por arquivo. Novo parametro flow_download_deployment, setado depois que o flow existe a partir do .fn.__name__ da propria funcao (nao repetido como string solta -- mesmo principio do Etapa(StrEnum)), permite nomes de variavel proprios por pipeline dentro do mesmo modulo. Tambem simplifica deploy_tags: tag da etapa vira so o nome dela (ex. "check_update"), sem o prefixo "etapa:" que so poluia.

CheckThenDownloadPipeline.run_download() agora chama upload_to_gcs ele mesmo (bucket_name="basedosdados-dev" fixo, convencao confirmada em ~42 datasets reais), usando data_path/dump_mode/source_format novos em DownloadResult -- download_data (ex-download_fn) so escreve o arquivo local e devolve onde ficou, sem repetir dataset_id/table_id/bucket_name em cada dataset. dump_mode/source_format continuam configuraveis (variam de verdade entre datasets: append/overwrite, csv/parquet). prefect_dataset_id passa a ser derivado automaticamente como f"{dataset_id}__{table_id}" -- mesma convencao ja usada em todo o repo real (br_bcb_agencia__agencia, br_denatran_frota__uf_tipo, br_rf_cno__{table_id}), aplicada mesmo com uma tabela so. Antes o default caia pra so dataset_id, o que colide entre tabelas de um dataset multi-tabela real -- exatamente o padrao que os dois pilotos de teste (ambos sob o mesmo dataset_id "test_dataset") ja expunham sem essa generalizacao. Constantes *_PREFECT_DATASET_ID viram redundantes e saem. Nome do @flow inverte pra "<etapa>: <dataset_id>" (etapa primeiro), centralizado em _flow_name() e exposto via novas properties check_update_flow_name/download_flow_name -- deployment_name() usa a mesma funcao, entao as duas pontas nunca divergem. Etapa.FLOW_DOWNLOAD -> Etapa.DOWNLOAD ("download"): a etapa ja era so "baixar e devolver o resultado" (o metodo correspondente sempre foi run_download, nunca run_flow_download) -- o valor "flow_download" era a unica coisa fora desse padrao. check_fn/download_fn -> check_for_update/download_data: o sufixo "_fn" generico nao aparece em mais nenhum lugar do repo, e nao dizia nada sobre o que cada callback faz -- os novos nomes descrevem a acao diretamente.

Aplica CheckThenDownloadPipeline (check_update -> download -> mat_test) em br_ans_beneficiario, br_denatran_frota, br_ibge_ipca (4 tabelas), br_inmet_bdmep, br_me_caged (3), br_me_cnpj (4), br_me_comex_stat (4), br_ms_cnes (13), br_sfb_sicar (9, ver ressalva) e us_cfpb_hmda -- 68 flows novos no total, zero colisao de nome (@flow ou deployment) entre eles. Todos os flows monoliticos antigos foram removidos dos respectivos flows.py (nao coexistem com os novos), mitigando o risco de um deploy pela branch de trabalho sobrescrever a fonte git de deployments reais em producao (ver stage_dispatch.py e a doc no ftwca). Terceira vez nesta issue que a contagem de tabelas do levantamento original estava errada (br_sfb_sicar: 9, nao 1) -- reforca a regra de sempre inspecionar o codigo real antes de migrar. stage_dispatch.py ganha `compare_against` (default "coverage", backward-compatible) em check_update_and_dispatch()/ CheckThenDownloadPipeline -- necessario pra br_me_cnpj.simples (NonHistorical, usa "table_update", valor que poll_source_for_update_task ja aceitava mas a capsula nao expunha). br_sfb_sicar foge do padrao genérico de proposito: as 9 tabelas sao materializadas e testadas juntas (dbt cruzado entre elas), o que o mat_test_flow generico nao suporta (1 tabela por vez) -- download e materializacao ficaram bundled num so flow, sem usar dispatch_mat_test. Precisa revisao humana antes de qualquer deploy real. br_bcb_sicor fica de fora, bloqueado e documentado, sem nenhum codigo alterado: compara tamanho em bytes da fonte, nao data, e a capsula so suporta comparacao por data hoje. Puramente preliminar: nada testado contra API/backend real, nada deployado no Prefect. Documentacao completa em ftwca/tasks/pipelines/issue_1867/migracao-lote-10-datasets.md.

mergify · 2026-09-09T20:19:06Z

@Winzen esse pull request tem conflitos 😩

Winzen changed the title ~~feat(test-event-pipeline): scaffold pilot flows for event-driven automations~~ feat: test-event-pipeline - scaffold pilot flows for event-driven automations Aug 30, 2026

Winzen self-assigned this Aug 31, 2026

Winzen and others added 24 commits August 31, 2026 19:27

[pre-commit.ci] auto fixes from pre-commit.com hooks

dbcf4c1

for more information, see https://pre-commit.ci

[pre-commit.ci] auto fixes from pre-commit.com hooks

5e779da

for more information, see https://pre-commit.ci

[pre-commit.ci] auto fixes from pre-commit.com hooks

852b029

for more information, see https://pre-commit.ci

[pre-commit.ci] auto fixes from pre-commit.com hooks

734e0b3

for more information, see https://pre-commit.ci

[pre-commit.ci] auto fixes from pre-commit.com hooks

b7f8e11

for more information, see https://pre-commit.ci

[pre-commit.ci] auto fixes from pre-commit.com hooks

cb06d8a

for more information, see https://pre-commit.ci

[pre-commit.ci] auto fixes from pre-commit.com hooks

8472abb

for more information, see https://pre-commit.ci

Winzen changed the title ~~feat: test-event-pipeline - scaffold pilot flows for event-driven automations~~ feat(1867): piloto de pipeline orientado a eventos (check_update -> flow_download -> mat_test via run_deployment()) Sep 1, 2026

Winzen changed the title ~~feat(1867): piloto de pipeline orientado a eventos (check_update -> flow_download -> mat_test via run_deployment())~~ feat: pipeline orientado a eventos (check_update -> flow_download -> mat_test) Sep 1, 2026

Winzen mentioned this pull request Sep 2, 2026

[chore] Pipeline orientado a eventos #1867

Open

5 tasks

Winzen added 4 commits September 2, 2026 22:49

Winzen mentioned this pull request Sep 3, 2026

bd.Table.create() só registra o ponteiro externo do BigQuery em basedosdados-staging, nunca em basedosdados-dev #1967

Open

laura-l-amaral linked an issue Sep 3, 2026 that may be closed by this pull request

[chore] Pipeline orientado a eventos #1867

Open

5 tasks

Winzen added 2 commits September 4, 2026 23:47

mergify Bot added the conflict [PR] Conflito de merge a resolver label Sep 9, 2026

Navigation Menu

Uh oh!

feat: pipeline orientado a eventos (check_update -> flow_download -> mat_test) - #1932

feat: pipeline orientado a eventos (check_update -> flow_download -> mat_test)#1932
Winzen wants to merge 31 commits into
mainfrom
feat/event-pipeline-automations-poc

Winzen commented Aug 30, 2026 •

edited

Loading

Uh oh!

coderabbitai Bot commented Aug 30, 2026

Draft PR not reviewed

Uh oh!

mergify Bot commented Sep 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

Winzen commented Aug 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Descrição do PR

Detalhes Técnicos

Teste e Validações

Riscos e Mitigações

Dependências

Uh oh!

coderabbitai Bot commented Aug 30, 2026

Draft PR not reviewed

Uh oh!

mergify Bot commented Sep 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Winzen commented Aug 30, 2026 •

edited

Loading