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.

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

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 30 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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Winzen commented Aug 30, 2026 •

edited

Loading