From 1c4f7b490ec5b2ff97521744bf5c1c251e1494e7 Mon Sep 17 00:00:00 2001
From: Pedro Castro <aspeddro@gmail.com>
Date: Fri, 7 Aug 2026 16:02:43 -0300
Subject: [PATCH 1/3] chore: adiciona Flow customizado com os atributos de
 deploy
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`deploy_flows.py` lê `deploy_schedules` e `job_variables` do objeto flow,
mas `prefect.Flow` não declara nenhum dos dois — atribuí-los funciona em
runtime e o Pyrefly acusa `missing-attribute`, o que vinha sendo silenciado
com 80 comentários `# pyrefly: ignore` espalhados pelos flows.

Adiciona `pipelines/utils/flow.py`: uma subclasse de `prefect.Flow` que
declara os dois atributos (vazios por padrão) e um decorator `flow` que a
instancia, com os mesmos argumentos do `prefect.flow`. Como herda de
`prefect.Flow`, as checagens `isinstance` do deploy e do próprio Prefect
seguem valendo — subclassear é o que o próprio Prefect faz em
`InfrastructureBoundFlow`.

Migra os 60 `flows.py` para `from pipelines.utils.flow import flow` e remove
os `# pyrefly: ignore [missing-attribute]` que existiam só por causa disso
(972 → 892 supressões, 0 diagnósticos). Em `deploy_flows.py`, `or None`
normaliza os padrões vazios para o que o Prefect recebia antes.

Verificação: o loader do CI descobre os mesmos 189 flows, com os mesmos
nomes, schedules e job_variables, antes e depois.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .claude/rules/prefect-pipeline-conventions.md |  21 +++
 .github/scripts/deploy_flows.py               |   7 +-
 AGENTS.md                                     |   9 +-
 pipelines/crawler/ibge_inflacao/flows.py      |   4 +-
 pipelines/datasets/au_abs_cpi/flows.py        |   4 +-
 .../datasets/au_abs_labour_force/flows.py     |   5 +-
 .../br_anatel_banda_larga_fixa/flows.py       |   4 +-
 .../br_anatel_telefonia_movel/flows.py        |   5 +-
 .../br_anp_precos_combustiveis/flows.py       |   4 +-
 .../datasets/br_ans_beneficiario/flows.py     |   4 +-
 pipelines/datasets/br_bcb_agencia/flows.py    |   4 +-
 pipelines/datasets/br_bcb_estban/flows.py     |   4 +-
 pipelines/datasets/br_bcb_sicor/flows.py      |   4 +-
 .../datasets/br_bcb_taxa_cambio/flows.py      |   4 +-
 pipelines/datasets/br_bcb_taxa_selic/flows.py |   4 +-
 pipelines/datasets/br_bd_indicadores/flows.py |  11 +-
 .../datasets/br_bd_siga_o_dinheiro/flows.py   |   4 +-
 .../br_bndes_operacoes_contratadas/flows.py   |   6 +-
 .../datasets/br_camara_dados_abertos/flows.py |   4 +-
 .../br_cgu_beneficios_cidadao/flows.py        |   4 +-
 .../datasets/br_cgu_cartao_pagamento/flows.py |   4 +-
 .../br_cgu_emendas_parlamentares/flows.py     |   4 +-
 .../br_cgu_licitacao_contrato/flows.py        |   4 +-
 .../br_cgu_pessoal_executivo_federal/flows.py |   4 +-
 .../flows.py                                  |   4 +-
 .../flows.py                                  |   4 +-
 .../br_cvm_administradores_carteira/flows.py  |   4 +-
 pipelines/datasets/br_cvm_fi/flows.py         |   4 +-
 .../flows.py                                  |   4 +-
 pipelines/datasets/br_denatran_frota/flows.py |   5 +-
 pipelines/datasets/br_fgv_igp/flows.py        |   4 +-
 pipelines/datasets/br_ibge_inpc/flows.py      |   4 +-
 pipelines/datasets/br_ibge_ipca/flows.py      |   4 +-
 pipelines/datasets/br_ibge_ipca15/flows.py    |   4 +-
 pipelines/datasets/br_ibge_pnadc/flows.py     |   5 +-
 pipelines/datasets/br_inmet_bdmep/flows.py    |   4 +-
 pipelines/datasets/br_me_caged/flows.py       |   4 +-
 pipelines/datasets/br_me_cnpj/flows.py        |   7 +-
 pipelines/datasets/br_me_comex_stat/flows.py  |   4 +-
 pipelines/datasets/br_me_rais/flows.py        |   5 +-
 pipelines/datasets/br_me_siconfi/flows.py     |   6 +-
 pipelines/datasets/br_mp_pep/flows.py         |   4 +-
 pipelines/datasets/br_ms_cnes/flows.py        |   4 +-
 pipelines/datasets/br_ms_sia/flows.py         |   4 +-
 pipelines/datasets/br_ms_sih/flows.py         |   4 +-
 pipelines/datasets/br_ms_sinan/flows.py       |   3 +-
 .../datasets/br_poder360_pesquisas/flows.py   |   4 +-
 pipelines/datasets/br_rf_cafir/flows.py       |   4 +-
 pipelines/datasets/br_rf_cno/flows.py         |   4 +-
 .../br_rj_isp_estatisticas_seguranca/flows.py |   4 +-
 .../datasets/br_senado_dados_abertos/flows.py |   5 +-
 pipelines/datasets/br_sfb_sicar/flows.py      |   4 +-
 .../datasets/br_stf_corte_aberta/flows.py     |   4 +-
 pipelines/datasets/br_tse_eleicoes/flows.py   |   4 +-
 pipelines/datasets/fundacao_lemann/flows.py   |   4 +-
 pipelines/datasets/test_dataset/flows.py      |   3 +-
 pipelines/datasets/us_bls_cpi/flows.py        |   5 +-
 pipelines/datasets/us_bls_qcew/flows.py       |   5 +-
 pipelines/datasets/world_cricsheet/flows.py   |   5 +-
 pipelines/utils/execute_dbt_model/flows.py    |   5 +-
 pipelines/utils/flow.py                       | 137 ++++++++++++++++++
 pipelines/utils/materialize_prod/flows.py     |   4 +-
 pipelines/utils/metadata/flows.py             |   4 +-
 pipelines/utils/tests/test_flow.py            |  58 ++++++++
 .../{{cookiecutter.pipeline_name}}/flows.py   |  11 +-
 65 files changed, 297 insertions(+), 204 deletions(-)
 create mode 100644 pipelines/utils/flow.py
 create mode 100644 pipelines/utils/tests/test_flow.py

diff --git a/.claude/rules/prefect-pipeline-conventions.md b/.claude/rules/prefect-pipeline-conventions.md
index eda694bd91..02b7230533 100644
--- a/.claude/rules/prefect-pipeline-conventions.md
+++ b/.claude/rules/prefect-pipeline-conventions.md
@@ -32,6 +32,20 @@ picks up `Flow` objects whose function is defined in that file
 (`obj.fn.__code__.co_filename` check). A factory that returns an inner `@flow`
 (see `br_ibge_ipca`) is fine because the inner fn is still defined in that file.
 
+### `@flow` comes from `pipelines.utils.flow`, not from `prefect`
+
+```python
+from pipelines.utils.flow import flow  # NOT `from prefect import flow`
+```
+
+The deploy script reads two attributes off the flow object — `deploy_schedules`
+and `job_variables` — that `prefect.Flow` does not declare, so setting them on a
+plain Prefect flow is a Pyrefly `missing-attribute` error. `pipelines/utils/flow.py`
+declares both on a `prefect.Flow` subclass and exports a `flow` decorator that
+builds it; it takes the same arguments as `prefect.flow` and the object stays a
+`prefect.Flow` for every `isinstance` check. Both attributes default to empty
+(no schedule, work-pool default infrastructure).
+
 ## DRY with the onboarding code
 
 The cleaning transform lives in **one place** and is shared:
@@ -275,6 +289,13 @@ on the **deployed Prefect worker** (its pod SA has access) — the local
 Schedule inline on the flow object (do NOT register storage/run-config by hand):
 
 ```python
+from pipelines.utils.flow import flow
+
+
+@flow(name="my_flow", log_prints=True)
+def my_flow() -> None: ...
+
+
 my_flow.deploy_schedules = [
     {"cron": "0 16 10,11,12,13 * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/.github/scripts/deploy_flows.py b/.github/scripts/deploy_flows.py
index 2b1af55d78..d685ae5267 100644
--- a/.github/scripts/deploy_flows.py
+++ b/.github/scripts/deploy_flows.py
@@ -69,7 +69,10 @@ def deploy_flow(
     entrypoint = f"{file_path}:{flow_name}"
     is_dev = "dev" in pool_name
 
-    schedules = getattr(flow, "deploy_schedules", None)
+    # `deploy_schedules` e `job_variables` são declarados em
+    # `pipelines.utils.flow.Flow` e vêm vazios quando o flow não os define;
+    # `or None` normaliza para o que o Prefect entende como "não informado".
+    schedules = getattr(flow, "deploy_schedules", None) or None
     if is_dev:
         schedules = None  # flows em dev não têm schedule
     elif schedules:
@@ -81,7 +84,7 @@ def deploy_flow(
             for s in schedules
         ]
 
-    job_variables = getattr(flow, "job_variables", None)
+    job_variables = getattr(flow, "job_variables", None) or None
 
     print(f"  Registrando {flow_name} → {entrypoint}")
 
diff --git a/AGENTS.md b/AGENTS.md
index 6372994909..e1cae5cd69 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -64,7 +64,7 @@ uv run manage.py add-pipeline <dataset_id>
 
 ### File conventions
 
-- `flows.py`: Define flows with `@flow`. Flows **must be defined at module level in this file** — `deploy_flows.py` only collects `Flow` objects whose function is defined there (an `obj.fn.__code__.co_filename` check).
+- `flows.py`: Define flows with `@flow` from **`pipelines.utils.flow`**, never `prefect.flow` — the repo's decorator returns a `prefect.Flow` subclass that declares the deploy attributes (`deploy_schedules`, `job_variables`), which the Prefect class does not, so setting them on a plain `prefect.Flow` is a Pyrefly `missing-attribute` error. Flows **must be defined at module level in this file** — `deploy_flows.py` only collects `Flow` objects whose function is defined there (an `obj.fn.__code__.co_filename` check).
 - `tasks.py`: Define tasks with `@task`.
 - `constants.py`: Use a `constants` enum or plain constants — no hardcoded values elsewhere.
 - `utils.py`: Pure helper functions with no Prefect decorators.
@@ -73,6 +73,13 @@ There is no `schedules.py`. Attach the schedule to the flow object in `flows.py`
 these dicts into `Cron` objects at deploy time:
 
 ```python
+from pipelines.utils.flow import flow
+
+
+@flow(name="my_flow", log_prints=True)
+def my_flow() -> None: ...
+
+
 my_flow.deploy_schedules = [
     {"cron": "0 16 10 * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/crawler/ibge_inflacao/flows.py b/pipelines/crawler/ibge_inflacao/flows.py
index 273f33ea9c..7cbda8ebac 100644
--- a/pipelines/crawler/ibge_inflacao/flows.py
+++ b/pipelines/crawler/ibge_inflacao/flows.py
@@ -3,13 +3,12 @@
 Prefect 3 — use os flows dos datasets (br_ibge_ipca, br_ibge_inpc) para deploy.
 """
 
-from prefect import flow
-
 from pipelines.crawler.ibge_inflacao.tasks import (
     check_for_updates,
     collect_data_utils,
     json_to_csv,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import DateFormat, PartBdpro, YearMonth
 from pipelines.utils.metadata.tasks import (
     commit_source_update_task,
@@ -136,5 +135,4 @@ def ibge_inflacao_flow(
     )
 
 
-# pyrefly: ignore [missing-attribute]
 ibge_inflacao_flow.deploy_schedules = []
diff --git a/pipelines/datasets/au_abs_cpi/flows.py b/pipelines/datasets/au_abs_cpi/flows.py
index 8f6329f750..d39eabbde1 100644
--- a/pipelines/datasets/au_abs_cpi/flows.py
+++ b/pipelines/datasets/au_abs_cpi/flows.py
@@ -15,10 +15,9 @@
 import shutil
 import tempfile
 
-from prefect import flow
-
 from pipelines.datasets.au_abs_cpi.constants import constants
 from pipelines.datasets.au_abs_cpi.tasks import clean_cpi, download_cpi
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     AllFree,
     DateFormat,
@@ -173,7 +172,6 @@ def au_abs_cpi_flow(
 # ABS publishes the monthly CPI in the last week of each month (moving to the
 # 4th Wednesday from Feb 2027). Poll across the last week at 16:00 BRT; the
 # source-poll guard no-ops until a new month lands.
-# pyrefly: ignore [missing-attribute]
 au_abs_cpi_flow.deploy_schedules = [
     {"cron": "0 16 22,23,24,25,26,27,28 * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/au_abs_labour_force/flows.py b/pipelines/datasets/au_abs_labour_force/flows.py
index ac7c428d40..9cd1858946 100644
--- a/pipelines/datasets/au_abs_labour_force/flows.py
+++ b/pipelines/datasets/au_abs_labour_force/flows.py
@@ -19,8 +19,6 @@
 import shutil
 import tempfile
 
-from prefect import flow
-
 from pipelines.datasets.au_abs_labour_force.constants import constants
 from pipelines.datasets.au_abs_labour_force.tasks import (
     clean_and_write_task,
@@ -28,6 +26,7 @@
     download_sdmx_task,
     latest_month_task,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     FreeLag,
@@ -191,10 +190,8 @@ def au_abs_labour_force_flow(
 # ABS releases Labour Force monthly, on a Thursday roughly the 3rd-4th week, at
 # 11:30 Canberra time. Poll daily across that window at 06:00 BRT (= evening AEST,
 # after the morning release); the source-poll guard no-ops until a new month lands.
-# pyrefly: ignore [missing-attribute]
 au_abs_labour_force_flow.deploy_schedules = [
     {"cron": "0 6 14-27 * *", "timezone": "America/Sao_Paulo"}
 ]
 # openpyxl reads the ~38 MB SEM1 pivot; give the worker headroom.
-# pyrefly: ignore [missing-attribute]
 au_abs_labour_force_flow.job_variables = {"memory": "6Gi"}
diff --git a/pipelines/datasets/br_anatel_banda_larga_fixa/flows.py b/pipelines/datasets/br_anatel_banda_larga_fixa/flows.py
index 33d3795dc1..a8e20d40a7 100644
--- a/pipelines/datasets/br_anatel_banda_larga_fixa/flows.py
+++ b/pipelines/datasets/br_anatel_banda_larga_fixa/flows.py
@@ -5,11 +5,10 @@
 deste diretório.
 """
 
-from prefect import flow
-
 from pipelines.crawler.anatel.banda_larga_fixa.flows import (
     _run_anatel_banda_larga_fixa,
 )
+from pipelines.utils.flow import flow
 
 
 def _anatel_blf_flow(table_id: str, cron: str | None):
@@ -39,7 +38,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = (
         [{"cron": cron, "timezone": "America/Sao_Paulo"}] if cron else []
     )
diff --git a/pipelines/datasets/br_anatel_telefonia_movel/flows.py b/pipelines/datasets/br_anatel_telefonia_movel/flows.py
index 7811fa7dc9..fcfda86327 100644
--- a/pipelines/datasets/br_anatel_telefonia_movel/flows.py
+++ b/pipelines/datasets/br_anatel_telefonia_movel/flows.py
@@ -1,10 +1,9 @@
 """Flows for br_anatel_telefonia_movel — Prefect 3."""
 
-from prefect import flow
-
 from pipelines.crawler.anatel.telefonia_movel.flows import (
     _run_anatel_telefonia_movel,
 )
+from pipelines.utils.flow import flow
 
 
 def _anatel_tm_flow(table_id: str, cron: str):
@@ -35,9 +34,7 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
-    # pyrefly: ignore [missing-attribute]
     _flow.job_variables = {"memory_limit": "8Gi", "memory_request": "2Gi"}
     return _flow
 
diff --git a/pipelines/datasets/br_anp_precos_combustiveis/flows.py b/pipelines/datasets/br_anp_precos_combustiveis/flows.py
index 9b82f3ddf3..bed3ed9703 100644
--- a/pipelines/datasets/br_anp_precos_combustiveis/flows.py
+++ b/pipelines/datasets/br_anp_precos_combustiveis/flows.py
@@ -2,13 +2,12 @@
 Flow br_anp_precos_combustiveis__microdados — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.anp_precos_combustiveis.tasks import (
     download_and_transform,
     get_data_source_anp_max_date,
     make_partitions,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     DateOnly,
@@ -120,7 +119,6 @@ def br_anp_precos_combustiveis__microdados(
             )
 
 
-# pyrefly: ignore [missing-attribute]
 br_anp_precos_combustiveis__microdados.deploy_schedules = [
     {"cron": "0 10 * * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_ans_beneficiario/flows.py b/pipelines/datasets/br_ans_beneficiario/flows.py
index 978ceb7c09..608578a9a8 100644
--- a/pipelines/datasets/br_ans_beneficiario/flows.py
+++ b/pipelines/datasets/br_ans_beneficiario/flows.py
@@ -2,14 +2,13 @@
 Flow br_ans_beneficiario__informacao_consolidada — Prefect 3.
 """
 
-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.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     PartBdpro,
@@ -133,7 +132,6 @@ def br_ans_beneficiario__informacao_consolidada(
             )
 
 
-# pyrefly: ignore [missing-attribute]
 br_ans_beneficiario__informacao_consolidada.deploy_schedules = [
     {"cron": "0 21 * * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_bcb_agencia/flows.py b/pipelines/datasets/br_bcb_agencia/flows.py
index 7777c39c5c..05c4edf759 100644
--- a/pipelines/datasets/br_bcb_agencia/flows.py
+++ b/pipelines/datasets/br_bcb_agencia/flows.py
@@ -2,8 +2,6 @@
 Flow br_bcb_agencia__agencia — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.bcb_agencia.tasks import (
     clean_data,
     download_table,
@@ -11,6 +9,7 @@
     get_documents_metadata,
     get_latest_file,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     PartBdpro,
@@ -146,7 +145,6 @@ def br_bcb_agencia__agencia(
             )
 
 
-# pyrefly: ignore [missing-attribute]
 br_bcb_agencia__agencia.deploy_schedules = [
     {"cron": "0 22 25-31 * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_bcb_estban/flows.py b/pipelines/datasets/br_bcb_estban/flows.py
index 5069b82474..19a80e85fb 100644
--- a/pipelines/datasets/br_bcb_estban/flows.py
+++ b/pipelines/datasets/br_bcb_estban/flows.py
@@ -2,8 +2,6 @@
 Flows para br_bcb_estban — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.bcb_estban.tasks import (
     cleaning_data,
     download_table,
@@ -12,6 +10,7 @@
     get_id_municipio,
     get_latest_file,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     PartBdpro,
@@ -167,7 +166,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_bcb_sicor/flows.py b/pipelines/datasets/br_bcb_sicor/flows.py
index c62ec69d99..4570c672a4 100644
--- a/pipelines/datasets/br_bcb_sicor/flows.py
+++ b/pipelines/datasets/br_bcb_sicor/flows.py
@@ -2,10 +2,9 @@
 Flows para br_bcb_sicor — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.bcb.flows import _run_bcb_sicor
 from pipelines.crawler.bcb.tasks import create_load_dictionary
+from pipelines.utils.flow import flow
 from pipelines.utils.tasks import (
     rename_flow_run_dataset_table,
     run_dbt,
@@ -52,7 +51,6 @@ def _flow(
             local_redis_execution=local_redis_execution,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_bcb_taxa_cambio/flows.py b/pipelines/datasets/br_bcb_taxa_cambio/flows.py
index f8694f21dc..42b2c47296 100644
--- a/pipelines/datasets/br_bcb_taxa_cambio/flows.py
+++ b/pipelines/datasets/br_bcb_taxa_cambio/flows.py
@@ -2,12 +2,11 @@
 Flow br_bcb_taxa_cambio — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.bcb_taxa_cambio.tasks import (
     get_data_taxa_cambio,
     treat_data_taxa_cambio,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     AllBdpro,
     DateFormat,
@@ -94,7 +93,6 @@ def br_bcb_taxa_cambio__taxa_cambio(
         )
 
 
-# pyrefly: ignore [missing-attribute]
 br_bcb_taxa_cambio__taxa_cambio.deploy_schedules = [
     {"cron": "0 8 * * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_bcb_taxa_selic/flows.py b/pipelines/datasets/br_bcb_taxa_selic/flows.py
index c9921f1fe7..1750a7b208 100644
--- a/pipelines/datasets/br_bcb_taxa_selic/flows.py
+++ b/pipelines/datasets/br_bcb_taxa_selic/flows.py
@@ -6,8 +6,9 @@
 
 import pandas as pd
 import requests
-from prefect import flow, task
+from prefect import task
 
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     AllBdpro,
     DateFormat,
@@ -154,7 +155,6 @@ def br_bcb_taxa_selic__taxa_selic(
             )
 
 
-# pyrefly: ignore [missing-attribute]
 br_bcb_taxa_selic__taxa_selic.deploy_schedules = [
     {"cron": "0 8 * * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_bd_indicadores/flows.py b/pipelines/datasets/br_bd_indicadores/flows.py
index 0cddc97877..224d70c455 100755
--- a/pipelines/datasets/br_bd_indicadores/flows.py
+++ b/pipelines/datasets/br_bd_indicadores/flows.py
@@ -2,8 +2,6 @@
 Flows for br_bd_indicadores — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.bd_indicadores.tasks import (
     crawler_metricas,
     crawler_real_time,
@@ -14,6 +12,7 @@
     has_new_tweets,
     save_data_to_csv,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.tasks import (
     download_data_to_gcs,
     rename_flow_run_dataset_table,
@@ -309,19 +308,11 @@ def br_bd_indicadores__pessoas(
 
 
 # Schedules — apenas contabilidade e receitas tinham schedule no Prefect 0
-# pyrefly: ignore [missing-attribute]
 br_bd_indicadores__contabilidade.deploy_schedules = []
-# pyrefly: ignore [missing-attribute]
 br_bd_indicadores__receitas_planejadas.deploy_schedules = []
-# pyrefly: ignore [missing-attribute]
 br_bd_indicadores__twitter_metrics.deploy_schedules = []
-# pyrefly: ignore [missing-attribute]
 br_bd_indicadores__twitter_metrics_agg.deploy_schedules = []
-# pyrefly: ignore [missing-attribute]
 br_bd_indicadores__page_views.deploy_schedules = []
-# pyrefly: ignore [missing-attribute]
 br_bd_indicadores__website_user.deploy_schedules = []
-# pyrefly: ignore [missing-attribute]
 br_bd_indicadores__equipes.deploy_schedules = []
-# pyrefly: ignore [missing-attribute]
 br_bd_indicadores__pessoas.deploy_schedules = []
diff --git a/pipelines/datasets/br_bd_siga_o_dinheiro/flows.py b/pipelines/datasets/br_bd_siga_o_dinheiro/flows.py
index fb65cc41ed..d257f8865d 100644
--- a/pipelines/datasets/br_bd_siga_o_dinheiro/flows.py
+++ b/pipelines/datasets/br_bd_siga_o_dinheiro/flows.py
@@ -2,9 +2,8 @@
 Flow br_bd_siga_o_dinheiro — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.bd_siga_o_dinheiro.tasks import get_table_ids
+from pipelines.utils.flow import flow
 from pipelines.utils.tasks import download_data_to_gcs, run_dbt
 
 
@@ -30,5 +29,4 @@ def br_bd_siga_o_dinheiro(
         download_data_to_gcs(dataset_id=dataset_id, table_id=table_id)
 
 
-# pyrefly: ignore [missing-attribute]
 br_bd_siga_o_dinheiro.deploy_schedules = []
diff --git a/pipelines/datasets/br_bndes_operacoes_contratadas/flows.py b/pipelines/datasets/br_bndes_operacoes_contratadas/flows.py
index a9d0eadd7b..eee6d1658a 100644
--- a/pipelines/datasets/br_bndes_operacoes_contratadas/flows.py
+++ b/pipelines/datasets/br_bndes_operacoes_contratadas/flows.py
@@ -5,12 +5,11 @@
 orquestracao (poll deferido) vive em pipelines/crawler/bndes/flows.py.
 """
 
-from prefect import flow
-
 from pipelines.crawler.bndes.flows import (
     _run_operacoes,
     _run_operacoes_administracao_publica,
 )
+from pipelines.utils.flow import flow
 
 
 @flow(
@@ -41,7 +40,6 @@ def br_bndes_operacoes_contratadas__operacoes_indiretas_automaticas(
     )
 
 
-# pyrefly: ignore [missing-attribute]
 br_bndes_operacoes_contratadas__operacoes_indiretas_automaticas.deploy_schedules = [
     {"cron": "0 6 * * 1", "timezone": "America/Sao_Paulo"}
 ]
@@ -75,7 +73,6 @@ def br_bndes_operacoes_contratadas__operacoes_nao_automaticas(
     )
 
 
-# pyrefly: ignore [missing-attribute]
 br_bndes_operacoes_contratadas__operacoes_nao_automaticas.deploy_schedules = [
     {"cron": "0 6 * * 1", "timezone": "America/Sao_Paulo"}
 ]
@@ -111,7 +108,6 @@ def br_bndes_operacoes_contratadas__operacoes_administracao_publica(
 
 # cron semanal (segunda 06h BRT), igual a irma; a fonte atualiza mensal e o poll
 # deferido no-opa quando nao ha novidade. Ajuste se quiser outra janela.
-# pyrefly: ignore [missing-attribute]
 br_bndes_operacoes_contratadas__operacoes_administracao_publica.deploy_schedules = [
     {"cron": "0 6 * * 1", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_camara_dados_abertos/flows.py b/pipelines/datasets/br_camara_dados_abertos/flows.py
index 62239e5203..0e63a0ad86 100644
--- a/pipelines/datasets/br_camara_dados_abertos/flows.py
+++ b/pipelines/datasets/br_camara_dados_abertos/flows.py
@@ -2,11 +2,10 @@
 Flows para br_camara_dados_abertos — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.camara_dados_abertos.flows import (
     _run_camara_dados_abertos,
 )
+from pipelines.utils.flow import flow
 
 
 def _camara_flow(table_id: str, cron: str):
@@ -33,7 +32,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_cgu_beneficios_cidadao/flows.py b/pipelines/datasets/br_cgu_beneficios_cidadao/flows.py
index f194ac6db3..78e616a3b3 100644
--- a/pipelines/datasets/br_cgu_beneficios_cidadao/flows.py
+++ b/pipelines/datasets/br_cgu_beneficios_cidadao/flows.py
@@ -6,9 +6,8 @@
 em pipelines/crawler/cgu/utils.py e pipelines/utils/utils.py.
 """
 
-from prefect import flow
-
 from pipelines.crawler.cgu.flows import _run_cgu_beneficios_cidadao
+from pipelines.utils.flow import flow
 
 
 def _flow_factory(table_id: str, cron: str):
@@ -37,7 +36,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_cgu_cartao_pagamento/flows.py b/pipelines/datasets/br_cgu_cartao_pagamento/flows.py
index 40f99b31d6..831230e0bf 100644
--- a/pipelines/datasets/br_cgu_cartao_pagamento/flows.py
+++ b/pipelines/datasets/br_cgu_cartao_pagamento/flows.py
@@ -2,9 +2,8 @@
 Flows para br_cgu_cartao_pagamento — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.cgu.flows import _run_cgu_cartao_pagamento
+from pipelines.utils.flow import flow
 
 
 def _flow_factory(table_id: str, cron: str):
@@ -33,7 +32,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_cgu_emendas_parlamentares/flows.py b/pipelines/datasets/br_cgu_emendas_parlamentares/flows.py
index 62905a438c..47b5c9c915 100644
--- a/pipelines/datasets/br_cgu_emendas_parlamentares/flows.py
+++ b/pipelines/datasets/br_cgu_emendas_parlamentares/flows.py
@@ -2,12 +2,11 @@
 Flows para br_cgu_emendas_parlamentares — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.cgu_emendas_parlamentares.tasks import (
     convert_str_to_float,
     get_last_modified_time,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     FreeLag,
@@ -119,7 +118,6 @@ def br_cgu_emendas_parlamentares__microdados(
             )
 
 
-# pyrefly: ignore [missing-attribute]
 br_cgu_emendas_parlamentares__microdados.deploy_schedules = [
     {"cron": "30 19 * * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_cgu_licitacao_contrato/flows.py b/pipelines/datasets/br_cgu_licitacao_contrato/flows.py
index a3c031164f..5538722191 100644
--- a/pipelines/datasets/br_cgu_licitacao_contrato/flows.py
+++ b/pipelines/datasets/br_cgu_licitacao_contrato/flows.py
@@ -2,9 +2,8 @@
 Flows para br_cgu_licitacao_contrato — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.cgu.flows import _run_cgu_licitacao_contrato
+from pipelines.utils.flow import flow
 
 
 def _flow_factory(table_id: str, cron: str | None):
@@ -34,7 +33,6 @@ def _flow(
         )
 
     if cron:
-        # pyrefly: ignore [missing-attribute]
         _flow.deploy_schedules = [
             {"cron": cron, "timezone": "America/Sao_Paulo"}
         ]
diff --git a/pipelines/datasets/br_cgu_pessoal_executivo_federal/flows.py b/pipelines/datasets/br_cgu_pessoal_executivo_federal/flows.py
index d3962bf117..e09d0809f0 100644
--- a/pipelines/datasets/br_cgu_pessoal_executivo_federal/flows.py
+++ b/pipelines/datasets/br_cgu_pessoal_executivo_federal/flows.py
@@ -1,11 +1,10 @@
 """Flows para br_cgu_pessoal_executivo_federal — Prefect 3."""
 
-from prefect import flow
-
 from pipelines.crawler.cgu_pessoal_executivo_federal.tasks import (
     clean_save_table,
     crawl,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.tasks import (
     rename_flow_run_dataset_table,
     run_dbt,
@@ -75,7 +74,6 @@ def br_cgu_pessoal_executivo_federal__terceirizados(
     )
 
 
-# pyrefly: ignore [missing-attribute]
 br_cgu_pessoal_executivo_federal__terceirizados.deploy_schedules = [
     {"cron": "0 0 28 2/4 *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_cgu_servidores_executivo_federal/flows.py b/pipelines/datasets/br_cgu_servidores_executivo_federal/flows.py
index 56160a3b21..71ec79abe6 100644
--- a/pipelines/datasets/br_cgu_servidores_executivo_federal/flows.py
+++ b/pipelines/datasets/br_cgu_servidores_executivo_federal/flows.py
@@ -17,9 +17,8 @@
 enquanto o ZIP é gerado de forma assíncrona).
 """
 
-from prefect import flow
-
 from pipelines.crawler.cgu.flows import _run_cgu_servidores_publicos
+from pipelines.utils.flow import flow
 
 
 def _flow_factory(table_id: str, cron: str):
@@ -48,7 +47,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_cnj_improbidade_administrativa/flows.py b/pipelines/datasets/br_cnj_improbidade_administrativa/flows.py
index 8c880a3ef7..4b9db6707f 100644
--- a/pipelines/datasets/br_cnj_improbidade_administrativa/flows.py
+++ b/pipelines/datasets/br_cnj_improbidade_administrativa/flows.py
@@ -2,14 +2,13 @@
 Flow br_cnj_improbidade_administrativa — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.cnj_improbidade_administrativa.tasks import (
     get_max_date,
     is_up_to_date,
     main_task,
     write_csv_file,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     DateOnly,
@@ -99,7 +98,6 @@ def br_cnj_improbidade_administrativa__condenacao(
         )
 
 
-# pyrefly: ignore [missing-attribute]
 br_cnj_improbidade_administrativa__condenacao.deploy_schedules = [
     {"cron": "0 7 * * 1", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_cvm_administradores_carteira/flows.py b/pipelines/datasets/br_cvm_administradores_carteira/flows.py
index 941da42c00..fa199e2c78 100644
--- a/pipelines/datasets/br_cvm_administradores_carteira/flows.py
+++ b/pipelines/datasets/br_cvm_administradores_carteira/flows.py
@@ -2,11 +2,10 @@
 Flows for br_cvm_administradores_carteira — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.cvm_administradores_carteira.flows import (
     _run_cvm_administradores_carteira,
 )
+from pipelines.utils.flow import flow
 
 
 def _adm_cart_flow(table_id: str, cron: str):
@@ -33,7 +32,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_cvm_fi/flows.py b/pipelines/datasets/br_cvm_fi/flows.py
index a66a1821fa..30d7a84e1e 100644
--- a/pipelines/datasets/br_cvm_fi/flows.py
+++ b/pipelines/datasets/br_cvm_fi/flows.py
@@ -2,9 +2,8 @@
 Flows for br_cvm_fi — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.cvm.flows import _run_cvm_fi
+from pipelines.utils.flow import flow
 
 
 def _cvm_fi_flow(table_id: str, cron: str, date_column_name: dict):
@@ -34,7 +33,6 @@ def _flow(
             url=url,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_cvm_oferta_publica_distribuicao/flows.py b/pipelines/datasets/br_cvm_oferta_publica_distribuicao/flows.py
index 3a0d7376ce..e8a4439963 100644
--- a/pipelines/datasets/br_cvm_oferta_publica_distribuicao/flows.py
+++ b/pipelines/datasets/br_cvm_oferta_publica_distribuicao/flows.py
@@ -2,12 +2,11 @@
 Flows for br_cvm_oferta_publica_distribuicao — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.cvm_oferta_publica_distribuicao.tasks import (
     clean_table_oferta_distribuicao,
     crawl,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     DateOnly,
@@ -95,7 +94,6 @@ def br_cvm_oferta_publica_distribuicao__dia(
         )
 
 
-# pyrefly: ignore [missing-attribute]
 br_cvm_oferta_publica_distribuicao__dia.deploy_schedules = [
     {"cron": "45 6 * * 1-5", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_denatran_frota/flows.py b/pipelines/datasets/br_denatran_frota/flows.py
index f42ea5bd75..f8e49ca1bd 100644
--- a/pipelines/datasets/br_denatran_frota/flows.py
+++ b/pipelines/datasets/br_denatran_frota/flows.py
@@ -2,8 +2,6 @@
 Flows para br_denatran_frota — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.denatran_frota.constants import (
     constants as denatran_constants,
 )
@@ -14,6 +12,7 @@
     treat_municipio_tipo_task,
     treat_uf_tipo_task,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     PartBdpro,
@@ -197,11 +196,9 @@ def br_denatran_frota__municipio_tipo(
     )
 
 
-# pyrefly: ignore [missing-attribute]
 br_denatran_frota__uf_tipo.deploy_schedules = [
     {"cron": "0 21 10-30 * *", "timezone": "America/Sao_Paulo"}
 ]
-# pyrefly: ignore [missing-attribute]
 br_denatran_frota__municipio_tipo.deploy_schedules = [
     {"cron": "20 21 10-30 * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_fgv_igp/flows.py b/pipelines/datasets/br_fgv_igp/flows.py
index 43b59ea067..f22776cf4c 100644
--- a/pipelines/datasets/br_fgv_igp/flows.py
+++ b/pipelines/datasets/br_fgv_igp/flows.py
@@ -2,9 +2,8 @@
 Flows for br_fgv_igp — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.fgv_igp.flows import _run_fgv_igp
+from pipelines.utils.flow import flow
 
 
 def _igp_flow(table_id: str, indice: str, periodo: str, cron: str | None):
@@ -33,7 +32,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = (
         [{"cron": cron, "timezone": "America/Sao_Paulo"}] if cron else []
     )
diff --git a/pipelines/datasets/br_ibge_inpc/flows.py b/pipelines/datasets/br_ibge_inpc/flows.py
index b7f351d69b..c501a1f119 100644
--- a/pipelines/datasets/br_ibge_inpc/flows.py
+++ b/pipelines/datasets/br_ibge_inpc/flows.py
@@ -2,9 +2,8 @@
 Flows para br_ibge_inpc — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.ibge_inflacao.flows import _run_ibge_inflacao
+from pipelines.utils.flow import flow
 
 
 def _inpc_flow(table_id: str, cron: str):
@@ -30,7 +29,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_ibge_ipca/flows.py b/pipelines/datasets/br_ibge_ipca/flows.py
index ca4b5fff11..a1d62a2d44 100644
--- a/pipelines/datasets/br_ibge_ipca/flows.py
+++ b/pipelines/datasets/br_ibge_ipca/flows.py
@@ -5,9 +5,8 @@
 (guards contra bloco vazio da API do IBGE) em utils.py.
 """
 
-from prefect import flow
-
 from pipelines.crawler.ibge_inflacao.flows import _run_ibge_inflacao
+from pipelines.utils.flow import flow
 
 
 def _ipca_flow(table_id: str, cron: str):
@@ -33,7 +32,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_ibge_ipca15/flows.py b/pipelines/datasets/br_ibge_ipca15/flows.py
index efad5a4d08..997bd1b6a9 100644
--- a/pipelines/datasets/br_ibge_ipca15/flows.py
+++ b/pipelines/datasets/br_ibge_ipca15/flows.py
@@ -2,9 +2,8 @@
 Flows para br_ibge_ipca15 — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.ibge_inflacao.flows import _run_ibge_inflacao
+from pipelines.utils.flow import flow
 
 
 def _ipca15_flow(table_id: str, cron: str):
@@ -33,7 +32,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_ibge_pnadc/flows.py b/pipelines/datasets/br_ibge_pnadc/flows.py
index 81dd45d004..f975915ee2 100644
--- a/pipelines/datasets/br_ibge_pnadc/flows.py
+++ b/pipelines/datasets/br_ibge_pnadc/flows.py
@@ -2,14 +2,13 @@
 Flow br_ibge_pnadc — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.ibge_pnadc.tasks import (
     build_partitions,
     build_table_paths,
     get_data_source_date_and_url,
 )
 from pipelines.datasets.br_ibge_pnadc.tasks import build_dicionario_task
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     PartBdpro,
@@ -121,7 +120,6 @@ def br_ibge_pnadc__microdados(
             )
 
 
-# pyrefly: ignore [missing-attribute]
 br_ibge_pnadc__microdados.deploy_schedules = [
     {"cron": "0 5 15-31 2,5,8,11 *", "timezone": "America/Sao_Paulo"}
 ]
@@ -196,7 +194,6 @@ def br_ibge_pnadc__dicionario(
     )
 
 
-# pyrefly: ignore [missing-attribute]
 br_ibge_pnadc__dicionario.deploy_schedules = [
     {"cron": "0 5 1,15 * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_inmet_bdmep/flows.py b/pipelines/datasets/br_inmet_bdmep/flows.py
index 2e0703b7b6..0ca882c000 100644
--- a/pipelines/datasets/br_inmet_bdmep/flows.py
+++ b/pipelines/datasets/br_inmet_bdmep/flows.py
@@ -2,12 +2,11 @@
 Flows for br_inmet_bdmep — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.inmet_bdmep.tasks import (
     extract_last_date_from_source,
     get_base_inmet,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     DateOnly,
@@ -115,7 +114,6 @@ def br_inmet_bdmep__microdados(
             )
 
 
-# pyrefly: ignore [missing-attribute]
 br_inmet_bdmep__microdados.deploy_schedules = [
     {"cron": "0 22 * * 1-5", "timezone": "America/Sao_Paulo"},
 ]
diff --git a/pipelines/datasets/br_me_caged/flows.py b/pipelines/datasets/br_me_caged/flows.py
index 1adb52087a..588bbc3f87 100644
--- a/pipelines/datasets/br_me_caged/flows.py
+++ b/pipelines/datasets/br_me_caged/flows.py
@@ -2,8 +2,6 @@
 Flows para br_me_caged — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.me_caged.tasks import (
     build_partitions,
     build_table_paths,
@@ -12,6 +10,7 @@
     get_source_last_date,
     get_table_last_date,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     PartBdpro,
@@ -149,7 +148,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_me_cnpj/flows.py b/pipelines/datasets/br_me_cnpj/flows.py
index 1c84aa3f29..a77ded7d6a 100644
--- a/pipelines/datasets/br_me_cnpj/flows.py
+++ b/pipelines/datasets/br_me_cnpj/flows.py
@@ -2,9 +2,8 @@
 Flows for br_me_cnpj — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.me_cnpj.flows import _run_me_cnpj
+from pipelines.utils.flow import flow
 
 
 def _me_cnpj_flow(table_id: str, cron: str):
@@ -31,20 +30,17 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
 
 br_me_cnpj__empresas = _me_cnpj_flow(table_id="empresas", cron="0 6 * * *")
-# pyrefly: ignore [missing-attribute]
 br_me_cnpj__empresas.job_variables = {
     "memory_limit": "5Gi",
     "memory_request": "2Gi",
 }
 
 br_me_cnpj__socios = _me_cnpj_flow(table_id="socios", cron="0 7 * * *")
-# pyrefly: ignore [missing-attribute]
 br_me_cnpj__socios.job_variables = {
     "memory_limit": "5Gi",
     "memory_request": "2Gi",
@@ -55,7 +51,6 @@ def _flow(
 br_me_cnpj__estabelecimentos = _me_cnpj_flow(
     table_id="estabelecimentos", cron="0 9 * * *"
 )
-# pyrefly: ignore [missing-attribute]
 br_me_cnpj__estabelecimentos.job_variables = {
     "memory_limit": "5Gi",
     "memory_request": "2Gi",
diff --git a/pipelines/datasets/br_me_comex_stat/flows.py b/pipelines/datasets/br_me_comex_stat/flows.py
index bf91b41afe..b7a05a975f 100644
--- a/pipelines/datasets/br_me_comex_stat/flows.py
+++ b/pipelines/datasets/br_me_comex_stat/flows.py
@@ -2,8 +2,6 @@
 Flows for br_me_comex_stat — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.me_comex_stat.constants import (
     constants as comex_constants,
 )
@@ -12,6 +10,7 @@
     download_br_me_comex_stat,
     parse_last_date,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     PartBdpro,
@@ -130,7 +129,6 @@ def _flow(
                     date_format="%Y-%m",
                 )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_me_rais/flows.py b/pipelines/datasets/br_me_rais/flows.py
index c87c132e6f..57d7455c45 100644
--- a/pipelines/datasets/br_me_rais/flows.py
+++ b/pipelines/datasets/br_me_rais/flows.py
@@ -2,9 +2,8 @@
 Flows for br_me_rais — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.me_rais.flows import _run_rais
+from pipelines.utils.flow import flow
 
 
 @flow(
@@ -61,7 +60,5 @@ def br_me_rais__microdados_vinculos(
     )
 
 
-# pyrefly: ignore [missing-attribute]
 br_me_rais__microdados_estabelecimentos.deploy_schedules = []
-# pyrefly: ignore [missing-attribute]
 br_me_rais__microdados_vinculos.deploy_schedules = []
diff --git a/pipelines/datasets/br_me_siconfi/flows.py b/pipelines/datasets/br_me_siconfi/flows.py
index de8b88e2f5..822ec312d8 100644
--- a/pipelines/datasets/br_me_siconfi/flows.py
+++ b/pipelines/datasets/br_me_siconfi/flows.py
@@ -23,10 +23,9 @@
 import tempfile
 from datetime import datetime
 
-from prefect import flow
-
 from pipelines.datasets.br_me_siconfi import tasks, utils
 from pipelines.datasets.br_me_siconfi.constants import constants
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import AllFree, DateFormat, YearOnly
 from pipelines.utils.metadata.tasks import (
     commit_source_update_task,
@@ -213,12 +212,10 @@ def br_me_siconfi_flow(
 
 # SICONFI is annual but revised retroactively; rebuild once a month (1st at
 # 16:00 BRT). Each run rebuilds fully — there is no source-poll no-op here.
-# pyrefly: ignore [missing-attribute]
 br_me_siconfi_flow.deploy_schedules = [
     {"cron": "0 16 1 * *", "timezone": "America/Sao_Paulo"}
 ]
 # The município window build holds a full year of data in pandas at a time.
-# pyrefly: ignore [missing-attribute]
 br_me_siconfi_flow.job_variables = {"memory": "16Gi"}
 
 
@@ -299,5 +296,4 @@ def br_me_siconfi_seed_flow(
 
 
 # The legacy build holds one year of município Excel in pandas at a time.
-# pyrefly: ignore [missing-attribute]
 br_me_siconfi_seed_flow.job_variables = {"memory": "8Gi"}
diff --git a/pipelines/datasets/br_mp_pep/flows.py b/pipelines/datasets/br_mp_pep/flows.py
index 020e4aecdf..cae2cebdcc 100644
--- a/pipelines/datasets/br_mp_pep/flows.py
+++ b/pipelines/datasets/br_mp_pep/flows.py
@@ -4,8 +4,6 @@
 
 import datetime
 
-from prefect import flow
-
 from pipelines.crawler.mp_pep.tasks import (
     clean_data,
     download_xlsx,
@@ -14,6 +12,7 @@
     scraper,
     setup_web_driver,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     PartBdpro,
@@ -112,7 +111,6 @@ def br_mp_pep__cargos_funcoes(
         )
 
 
-# pyrefly: ignore [missing-attribute]
 br_mp_pep__cargos_funcoes.deploy_schedules = [
     {"cron": "0 14 * * 3", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_ms_cnes/flows.py b/pipelines/datasets/br_ms_cnes/flows.py
index d78adc729e..f8acd40c09 100644
--- a/pipelines/datasets/br_ms_cnes/flows.py
+++ b/pipelines/datasets/br_ms_cnes/flows.py
@@ -13,9 +13,8 @@
 este arquivo, o deploy sai "0 registrados, N pulados" e passa.
 """
 
-from prefect import flow
-
 from pipelines.crawler.datasus.flows import _run_cnes
+from pipelines.utils.flow import flow
 
 
 def _cnes_flow(table_id: str, cron: str | None):
@@ -45,7 +44,6 @@ def _flow(
         )
 
     if cron:
-        # pyrefly: ignore [missing-attribute]
         _flow.deploy_schedules = [
             {"cron": cron, "timezone": "America/Sao_Paulo"}
         ]
diff --git a/pipelines/datasets/br_ms_sia/flows.py b/pipelines/datasets/br_ms_sia/flows.py
index d33651e882..30e7b0cc56 100644
--- a/pipelines/datasets/br_ms_sia/flows.py
+++ b/pipelines/datasets/br_ms_sia/flows.py
@@ -2,9 +2,8 @@
 Flows for br_ms_sia — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.datasus.flows import _run_siasus
+from pipelines.utils.flow import flow
 
 
 def _sia_flow(table_id: str, cron: str):
@@ -33,7 +32,6 @@ def _flow(
             year_month_to_extract=year_month_to_extract,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_ms_sih/flows.py b/pipelines/datasets/br_ms_sih/flows.py
index b1a72dbf2f..a21787f2f4 100644
--- a/pipelines/datasets/br_ms_sih/flows.py
+++ b/pipelines/datasets/br_ms_sih/flows.py
@@ -2,9 +2,8 @@
 Flows for br_ms_sih — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.datasus.flows import _run_sihsus
+from pipelines.utils.flow import flow
 
 
 def _sih_flow(table_id: str, cron: str):
@@ -33,7 +32,6 @@ def _flow(
             year_month_to_extract=year_month_to_extract,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_ms_sinan/flows.py b/pipelines/datasets/br_ms_sinan/flows.py
index 1ecade2593..6c010104a8 100644
--- a/pipelines/datasets/br_ms_sinan/flows.py
+++ b/pipelines/datasets/br_ms_sinan/flows.py
@@ -2,9 +2,8 @@
 Flows for br_ms_sinan — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.datasus.flows import _run_sinan
+from pipelines.utils.flow import flow
 
 
 @flow(
diff --git a/pipelines/datasets/br_poder360_pesquisas/flows.py b/pipelines/datasets/br_poder360_pesquisas/flows.py
index 73fe414b75..81e02b1c02 100644
--- a/pipelines/datasets/br_poder360_pesquisas/flows.py
+++ b/pipelines/datasets/br_poder360_pesquisas/flows.py
@@ -2,9 +2,8 @@
 Flow br_poder360_pesquisas — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.poder360_pesquisas.tasks import crawler
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     DateOnly,
@@ -88,7 +87,6 @@ def br_poder360_pesquisas__microdados(
         )
 
 
-# pyrefly: ignore [missing-attribute]
 br_poder360_pesquisas__microdados.deploy_schedules = [
     {"cron": "42 3 * * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_rf_cafir/flows.py b/pipelines/datasets/br_rf_cafir/flows.py
index 350239fad4..6074c62d2d 100644
--- a/pipelines/datasets/br_rf_cafir/flows.py
+++ b/pipelines/datasets/br_rf_cafir/flows.py
@@ -2,8 +2,6 @@
 Flows for br_rf_cafir — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.rf_cafir.constants import (
     constants as br_rf_cafir_constants,
 )
@@ -13,6 +11,7 @@
     task_get_last_update_date,
     task_parse_api_metadata,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     DateOnly,
@@ -134,7 +133,6 @@ def br_rf_cafir__imoveis_rurais(
             )
 
 
-# pyrefly: ignore [missing-attribute]
 br_rf_cafir__imoveis_rurais.deploy_schedules = [
     {"cron": "0 0 * * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_rf_cno/flows.py b/pipelines/datasets/br_rf_cno/flows.py
index 5d0c5306af..17ab200cf9 100644
--- a/pipelines/datasets/br_rf_cno/flows.py
+++ b/pipelines/datasets/br_rf_cno/flows.py
@@ -9,9 +9,8 @@
 `safe_cast(data as date)` do model virava NULL e o filtro incremental nunca inseria.
 """
 
-from prefect import flow
-
 from pipelines.crawler.rf.flows import _run_rf
+from pipelines.utils.flow import flow
 
 
 def _cno_flow(table_id: str, cron: str):
@@ -40,7 +39,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_rj_isp_estatisticas_seguranca/flows.py b/pipelines/datasets/br_rj_isp_estatisticas_seguranca/flows.py
index ea6b602a94..d7fa71b929 100644
--- a/pipelines/datasets/br_rj_isp_estatisticas_seguranca/flows.py
+++ b/pipelines/datasets/br_rj_isp_estatisticas_seguranca/flows.py
@@ -2,9 +2,8 @@
 Flows for br_rj_isp_estatisticas_seguranca — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.isp.flows import _run_isp
+from pipelines.utils.flow import flow
 
 
 def _isp_flow(table_id: str, cron: str):
@@ -31,7 +30,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
     return _flow
 
diff --git a/pipelines/datasets/br_senado_dados_abertos/flows.py b/pipelines/datasets/br_senado_dados_abertos/flows.py
index 4ecc255cb7..dfad7b2c59 100644
--- a/pipelines/datasets/br_senado_dados_abertos/flows.py
+++ b/pipelines/datasets/br_senado_dados_abertos/flows.py
@@ -16,10 +16,9 @@
 import shutil
 import tempfile
 
-from prefect import flow
-
 from pipelines.datasets.br_senado_dados_abertos.constants import constants
 from pipelines.datasets.br_senado_dados_abertos.tasks import extract_clean
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     DateOnly,
@@ -165,9 +164,7 @@ def br_senado_dados_abertos_flow(
 
 
 # Legislative activity updates on business days; refresh every morning (BRT).
-# pyrefly: ignore [missing-attribute]
 br_senado_dados_abertos_flow.deploy_schedules = [
     {"cron": "0 8 * * *", "timezone": "America/Sao_Paulo"}
 ]
-# pyrefly: ignore [missing-attribute]
 br_senado_dados_abertos_flow.job_variables = {"memory": "4Gi"}
diff --git a/pipelines/datasets/br_sfb_sicar/flows.py b/pipelines/datasets/br_sfb_sicar/flows.py
index b3dc147e4c..4e2c8b8f93 100644
--- a/pipelines/datasets/br_sfb_sicar/flows.py
+++ b/pipelines/datasets/br_sfb_sicar/flows.py
@@ -2,14 +2,13 @@
 Flow br_sfb_sicar — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.sfb_sicar.constants import Constants
 from pipelines.crawler.sfb_sicar.tasks import (
     download_car,
     get_each_uf_release_date,
     unzip_to_parquet,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     DateOnly,
@@ -113,7 +112,6 @@ def br_sfb_sicar__area_imovel(
         )
 
 
-# pyrefly: ignore [missing-attribute]
 br_sfb_sicar__area_imovel.deploy_schedules = [
     {"cron": "15 21 15 * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_stf_corte_aberta/flows.py b/pipelines/datasets/br_stf_corte_aberta/flows.py
index cc10eb4e45..0106ef149a 100644
--- a/pipelines/datasets/br_stf_corte_aberta/flows.py
+++ b/pipelines/datasets/br_stf_corte_aberta/flows.py
@@ -2,13 +2,12 @@
 Flow br_stf_corte_aberta — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.stf_corte_aberta.tasks import (
     download_and_transform,
     get_data_source_stf_max_date,
     make_partitions,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     DateOnly,
@@ -119,7 +118,6 @@ def br_stf_corte_aberta__decisoes(
             )
 
 
-# pyrefly: ignore [missing-attribute]
 br_stf_corte_aberta__decisoes.deploy_schedules = [
     {"cron": "0 12 * * *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/br_tse_eleicoes/flows.py b/pipelines/datasets/br_tse_eleicoes/flows.py
index 1512902939..a1a1815cdb 100644
--- a/pipelines/datasets/br_tse_eleicoes/flows.py
+++ b/pipelines/datasets/br_tse_eleicoes/flows.py
@@ -2,9 +2,8 @@
 Flows for br_tse_eleicoes — Prefect 3.
 """
 
-from prefect import flow
-
 from pipelines.crawler.tse_eleicoes.flows import _run_tse_eleicoes
+from pipelines.utils.flow import flow
 
 
 def _tse_flow(table_id: str, cron: str | None):
@@ -31,7 +30,6 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = (
         [{"cron": cron, "timezone": "America/Sao_Paulo"}] if cron else []
     )
diff --git a/pipelines/datasets/fundacao_lemann/flows.py b/pipelines/datasets/fundacao_lemann/flows.py
index ffa8044fcf..eac9baf956 100644
--- a/pipelines/datasets/fundacao_lemann/flows.py
+++ b/pipelines/datasets/fundacao_lemann/flows.py
@@ -2,8 +2,7 @@
 Flow fundacao_lemann — Prefect 3.
 """
 
-from prefect import flow
-
+from pipelines.utils.flow import flow
 from pipelines.utils.tasks import (
     download_data_to_gcs,
     rename_flow_run_dataset_table,
@@ -40,7 +39,6 @@ def fundacao_lemann__ano_escola_serie_educacao_aprendizagem_adequada(
     download_data_to_gcs(dataset_id=dataset_id, table_id=table_id)
 
 
-# pyrefly: ignore [missing-attribute]
 fundacao_lemann__ano_escola_serie_educacao_aprendizagem_adequada.deploy_schedules = [
     {"cron": "0 9 1 1 *", "timezone": "America/Sao_Paulo"}
 ]
diff --git a/pipelines/datasets/test_dataset/flows.py b/pipelines/datasets/test_dataset/flows.py
index 3d0740fefb..c2ea2101f2 100644
--- a/pipelines/datasets/test_dataset/flows.py
+++ b/pipelines/datasets/test_dataset/flows.py
@@ -2,8 +2,7 @@
 Flows de teste para a função download_data_to_gcs.
 """
 
-from prefect import flow
-
+from pipelines.utils.flow import flow
 from pipelines.utils.tasks import run_dbt
 
 DATASET_ID = "test_dataset"
diff --git a/pipelines/datasets/us_bls_cpi/flows.py b/pipelines/datasets/us_bls_cpi/flows.py
index 04c57b0616..4e9b3721f1 100644
--- a/pipelines/datasets/us_bls_cpi/flows.py
+++ b/pipelines/datasets/us_bls_cpi/flows.py
@@ -13,10 +13,9 @@
 import shutil
 import tempfile
 
-from prefect import flow
-
 from pipelines.datasets.us_bls_cpi.constants import constants
 from pipelines.datasets.us_bls_cpi.tasks import clean_cpi, download_cpi
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     AllFree,
     DateFormat,
@@ -177,10 +176,8 @@ def us_bls_cpi_flow(
 # BLS releases CPI monthly, ~2nd week, on a US business day. Poll across a few
 # mid-month days at 16:00 BRT; the source-poll guard no-ops until a new month
 # actually appears.
-# pyrefly: ignore [missing-attribute]
 us_bls_cpi_flow.deploy_schedules = [
     {"cron": "0 16 10,11,12,13,14,15 * *", "timezone": "America/Sao_Paulo"}
 ]
 # The clean step holds ~4M rows in pandas; give the worker headroom.
-# pyrefly: ignore [missing-attribute]
 us_bls_cpi_flow.job_variables = {"memory": "8Gi"}
diff --git a/pipelines/datasets/us_bls_qcew/flows.py b/pipelines/datasets/us_bls_qcew/flows.py
index 3c1dc69f2d..d1cd883a01 100644
--- a/pipelines/datasets/us_bls_qcew/flows.py
+++ b/pipelines/datasets/us_bls_qcew/flows.py
@@ -19,13 +19,12 @@
 import shutil
 import tempfile
 
-from prefect import flow
-
 from pipelines.datasets.us_bls_qcew.constants import constants
 from pipelines.datasets.us_bls_qcew.tasks import (
     clean_qcew,
     latest_source_period,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     AllFree,
     DateFormat,
@@ -188,11 +187,9 @@ def us_bls_qcew_flow(
 # and Wages news releases land in early March, June, September, and December.
 # Poll across the first ~10 days of those months at 16:00 BRT; the source-poll
 # guard no-ops until a new quarter actually appears in the singlefiles.
-# pyrefly: ignore [missing-attribute]
 us_bls_qcew_flow.deploy_schedules = [
     {"cron": "0 16 1-10 3,6,9,12 *", "timezone": "America/Sao_Paulo"}
 ]
 # The clean step streams ~15M-row singlefiles one chunk at a time (peak ~1.75GB
 # in pandas); give the worker headroom above that.
-# pyrefly: ignore [missing-attribute]
 us_bls_qcew_flow.job_variables = {"memory": "8Gi"}
diff --git a/pipelines/datasets/world_cricsheet/flows.py b/pipelines/datasets/world_cricsheet/flows.py
index 56da898510..bf0f09cdef 100644
--- a/pipelines/datasets/world_cricsheet/flows.py
+++ b/pipelines/datasets/world_cricsheet/flows.py
@@ -16,13 +16,12 @@
 import shutil
 import tempfile
 
-from prefect import flow
-
 from pipelines.datasets.world_cricsheet.constants import constants
 from pipelines.datasets.world_cricsheet.tasks import (
     clean_cricsheet,
     download_cricsheet,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     AllFree,
     DateFormat,
@@ -212,11 +211,9 @@ def world_cricsheet_flow(
 # WEEKLY instead — Monday 06:00 BRT — which cuts the cost ~4x while keeping
 # freshness fine. The source-poll guard still no-ops between real releases, and
 # the full-replace dump means overlapping windows never duplicate.
-# pyrefly: ignore [missing-attribute]
 world_cricsheet_flow.deploy_schedules = [
     {"cron": "0 6 * * 1", "timezone": "America/Sao_Paulo"}
 ]
 # The deliveries build streams 11.4M rows and the bundle extracts to several GB;
 # give the worker headroom.
-# pyrefly: ignore [missing-attribute]
 world_cricsheet_flow.job_variables = {"memory": "12Gi"}
diff --git a/pipelines/utils/execute_dbt_model/flows.py b/pipelines/utils/execute_dbt_model/flows.py
index 70c4c57931..b8388c9d08 100644
--- a/pipelines/utils/execute_dbt_model/flows.py
+++ b/pipelines/utils/execute_dbt_model/flows.py
@@ -9,7 +9,9 @@
 from typing import Any
 
 from dbt.cli.main import dbtRunner
-from prefect import flow, task
+from prefect import task
+
+from pipelines.utils.flow import flow
 
 
 @task
@@ -115,5 +117,4 @@ def run_dbt_model_flow(
         dbt_done.result()
 
 
-# pyrefly: ignore [missing-attribute]
 run_dbt_model_flow.deploy_schedules = []
diff --git a/pipelines/utils/flow.py b/pipelines/utils/flow.py
new file mode 100644
index 0000000000..33841e654a
--- /dev/null
+++ b/pipelines/utils/flow.py
@@ -0,0 +1,137 @@
+"""
+`Flow` da Base dos Dados — o `Flow` do Prefect 3 mais os atributos de deploy.
+
+`.github/scripts/deploy_flows.py` lê dois atributos do objeto flow que o
+`prefect.Flow` não declara:
+
+- `deploy_schedules`: lista de agendamentos cron, convertida em objetos
+  `prefect.schedules.Cron` no deploy de produção (no pool de dev os schedules
+  são descartados);
+- `job_variables`: overrides da configuração de infraestrutura do work pool
+  (memória, CPU...).
+
+Atribuí-los a um `prefect.Flow` funciona em runtime, mas o Pyrefly acusa
+`missing-attribute`, já que a classe do Prefect não os declara. Este módulo
+declara os dois numa subclasse de `prefect.Flow` e expõe um decorator `flow`
+que instancia essa subclasse. Use sempre este `flow` em `flows.py`, nunca o
+`prefect.flow`:
+
+```python
+from pipelines.utils.flow import flow
+
+
+@flow(name="meu_dataset", log_prints=True)
+def meu_dataset_flow() -> None: ...
+
+
+meu_dataset_flow.deploy_schedules = [
+    {"cron": "0 16 10 * *", "timezone": "America/Sao_Paulo"}
+]
+meu_dataset_flow.job_variables = {"memory": "8Gi"}
+```
+
+Como `Flow` herda de `prefect.Flow`, as checagens `isinstance(obj, Flow)` do
+script de deploy e do próprio Prefect continuam valendo.
+"""
+
+from collections.abc import Callable
+from typing import (
+    Any,
+    NotRequired,
+    ParamSpec,
+    TypedDict,
+    TypeVar,
+    overload,
+)
+
+from prefect import Flow as PrefectFlow
+from prefect.futures import PrefectFuture
+from prefect.task_runners import TaskRunner
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+class DeploySchedule(TypedDict):
+    """Agendamento de um flow, no formato lido por `deploy_flows.py`.
+
+    Attributes:
+        cron: Expressão cron (ver crontab.guru).
+        timezone: Fuso da expressão; `"UTC"` quando omitido. Por convenção do
+            repositório, use `"America/Sao_Paulo"`.
+    """
+
+    cron: str
+    timezone: NotRequired[str]
+
+
+class Flow(PrefectFlow[P, R]):
+    """Flow do Prefect 3 com os atributos que o deploy da BD lê.
+
+    Attributes:
+        deploy_schedules: Agendamentos do flow. Vazio (o padrão) significa
+            deployment sem schedule — execução apenas manual.
+        job_variables: Overrides da configuração de infraestrutura do work
+            pool, por exemplo `{"memory": "8Gi"}`. Vazio usa o padrão do pool.
+    """
+
+    deploy_schedules: list[DeploySchedule]
+    job_variables: dict[str, Any]
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.deploy_schedules = []
+        self.job_variables = {}
+
+
+# Uso sem parênteses (`@flow`). A assinatura da função decorada não é
+# preservada aqui — usar `Callable[P, R]` tornaria esta sobrecarga genérica, o
+# que exigiria a sintaxe de type parameters do Python 3.12 (ruff UP047).
+@overload
+def flow(fn: Callable[..., Any], /) -> Flow[..., Any]: ...
+
+
+@overload
+def flow(
+    fn: None = None,
+    /,
+    *,
+    name: str | None = None,
+    version: str | None = None,
+    flow_run_name: Callable[[], str] | str | None = None,
+    retries: int | None = None,
+    retry_delay_seconds: int | float | None = None,
+    task_runner: TaskRunner[PrefectFuture[Any]] | None = None,
+    description: str | None = None,
+    timeout_seconds: int | float | None = None,
+    validate_parameters: bool = True,
+    persist_result: bool | None = None,
+    cache_result_in_memory: bool = True,
+    log_prints: bool | None = None,
+    **kwargs: Any,
+) -> Callable[[Callable[P, R]], Flow[P, R]]: ...
+
+
+def flow(
+    fn: Callable[..., Any] | None = None, /, **kwargs: Any
+) -> Flow[..., Any] | Callable[[Callable[..., Any]], Flow[..., Any]]:
+    """Decorator equivalente ao `prefect.flow`, mas devolvendo um `Flow` da BD.
+
+    Aceita os mesmos argumentos nomeados do `prefect.flow` (`name`,
+    `log_prints`, `retries`, `flow_run_name`, ...), repassados ao construtor do
+    `prefect.Flow`.
+
+    Args:
+        fn: A função decorada, quando o decorator é usado sem parênteses.
+        **kwargs: Argumentos do `prefect.flow`.
+
+    Returns:
+        O `Flow` construído, ou o decorator que o constrói.
+    """
+    if fn is not None:
+        return Flow(fn=fn, **kwargs)
+
+    def decorator(fn: Callable[..., Any]) -> Flow[..., Any]:
+        return Flow(fn=fn, **kwargs)
+
+    return decorator
diff --git a/pipelines/utils/materialize_prod/flows.py b/pipelines/utils/materialize_prod/flows.py
index fdc7a17d1b..f53eb8519d 100644
--- a/pipelines/utils/materialize_prod/flows.py
+++ b/pipelines/utils/materialize_prod/flows.py
@@ -5,8 +5,7 @@
 
 from __future__ import annotations
 
-from prefect import flow
-
+from pipelines.utils.flow import flow
 from pipelines.utils.materialize_prod.tasks import (
     download_files_from_bucket_folders,
 )
@@ -170,5 +169,4 @@ def transfer_files_to_prod_flow(
         )
 
 
-# pyrefly: ignore [missing-attribute]
 transfer_files_to_prod_flow.deploy_schedules = []  # utilitário, disparo manual
diff --git a/pipelines/utils/metadata/flows.py b/pipelines/utils/metadata/flows.py
index 76508f2ccd..2d6a9ebf9c 100644
--- a/pipelines/utils/metadata/flows.py
+++ b/pipelines/utils/metadata/flows.py
@@ -9,8 +9,7 @@
 Pydantic no parsing do parâmetro do flow.
 """
 
-from prefect import flow
-
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import CoverageSpec
 from pipelines.utils.metadata.tasks import (
     register_table_materialization_task,
@@ -37,5 +36,4 @@ def update_temporal_coverage(
     )
 
 
-# pyrefly: ignore [missing-attribute]
 update_temporal_coverage.deploy_schedules = []
diff --git a/pipelines/utils/tests/test_flow.py b/pipelines/utils/tests/test_flow.py
new file mode 100644
index 0000000000..8ac628433c
--- /dev/null
+++ b/pipelines/utils/tests/test_flow.py
@@ -0,0 +1,58 @@
+"""Testes do decorator `flow` da BD (`pipelines.utils.flow`)."""
+
+from prefect import Flow as PrefectFlow
+
+from pipelines.utils.flow import DeploySchedule, Flow, flow
+
+
+@flow(name="flow_de_teste", log_prints=True)
+def _flow_com_opcoes(x: int = 1) -> int:
+    return x
+
+
+@flow
+def _flow_sem_parenteses() -> str:
+    return "ok"
+
+
+def test_e_um_flow_do_prefect():
+    """O deploy e o próprio Prefect fazem `isinstance(obj, prefect.Flow)`."""
+    assert isinstance(_flow_com_opcoes, Flow)
+    assert isinstance(_flow_com_opcoes, PrefectFlow)
+    assert isinstance(_flow_sem_parenteses, PrefectFlow)
+
+
+def test_repassa_as_opcoes_do_prefect():
+    assert _flow_com_opcoes.name == "flow_de_teste"
+    assert _flow_com_opcoes.log_prints is True
+    # Sem `name=`, o Prefect infere o nome a partir da função.
+    assert _flow_sem_parenteses.name == "-flow-sem-parenteses"
+
+
+def test_atributos_de_deploy_comecam_vazios():
+    assert _flow_com_opcoes.deploy_schedules == []
+    assert _flow_com_opcoes.job_variables == {}
+
+
+def test_atributos_de_deploy_nao_sao_compartilhados():
+    """Cada flow tem os seus — nada de default mutável de classe."""
+    assert (
+        _flow_com_opcoes.deploy_schedules
+        is not _flow_sem_parenteses.deploy_schedules
+    )
+    assert (
+        _flow_com_opcoes.job_variables
+        is not _flow_sem_parenteses.job_variables
+    )
+
+
+def test_aceita_atribuicao_dos_atributos_de_deploy():
+    schedules: list[DeploySchedule] = [
+        {"cron": "0 16 10 * *", "timezone": "America/Sao_Paulo"}
+    ]
+
+    _flow_com_opcoes.deploy_schedules = schedules
+    _flow_com_opcoes.job_variables = {"memory": "8Gi"}
+
+    assert _flow_com_opcoes.deploy_schedules == schedules
+    assert _flow_com_opcoes.job_variables == {"memory": "8Gi"}
diff --git a/pipelines/{{cookiecutter.pipeline_name}}/flows.py b/pipelines/{{cookiecutter.pipeline_name}}/flows.py
index da266dc06b..986db05071 100644
--- a/pipelines/{{cookiecutter.pipeline_name}}/flows.py
+++ b/pipelines/{{cookiecutter.pipeline_name}}/flows.py
@@ -6,9 +6,11 @@
 #
 # Aqui é onde devem ser definidos os flows da pipeline (Prefect 3).
 #
-# Cada flow é uma função decorada com `@flow`. Os passos são chamadas de
-# tasks (funções decoradas com `@task`, definidas em `tasks.py`) executadas
-# na ordem do corpo da função.
+# Cada flow é uma função decorada com `@flow` — o decorator da BD, em
+# `pipelines.utils.flow`, e não o do Prefect: ele devolve um `Flow` que declara
+# os atributos de deploy (`deploy_schedules`, `job_variables`). Os passos são
+# chamadas de tasks (funções decoradas com `@task`, definidas em `tasks.py`)
+# executadas na ordem do corpo da função.
 #
 # O deploy é feito por `.github/scripts/deploy_flows.py`, que descobre os
 # objetos `Flow` deste arquivo automaticamente — não é preciso registrar
@@ -25,9 +27,8 @@
 #
 ###############################################################################
 
-from prefect import flow
-
 from pipelines.datasets.{{cookiecutter.pipeline_name}}.tasks import say_hello
+from pipelines.utils.flow import flow
 
 
 @flow(name="{{cookiecutter.pipeline_name}}", log_prints=True)

From 3b3142f2196bd0be5d593534b82ad2b9520ead85 Mon Sep 17 00:00:00 2001
From: Pedro Castro <aspeddro@gmail.com>
Date: Fri, 7 Aug 2026 21:43:10 -0300
Subject: [PATCH 2/3] chore: usa Cron do prefect.schedules em deploy_schedules
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Troca os dicts `{"cron": ..., "timezone": ...}` pelo `Cron` do
`prefect.schedules`, que já recebe o `timezone` — 54 agendamentos em 50
`flows.py`. `deploy_schedules` passa a ser `list[Schedule]` (o que o `Cron`
devolve) e `deploy_flows.py` não precisa mais converter dict → Cron.

Os agendamentos agora são validados na importação do módulo, e não só no
deploy.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .claude/rules/prefect-pipeline-conventions.md |  9 +++--
 .github/scripts/deploy_flows.py               |  9 -----
 AGENTS.md                                     | 10 ++---
 pipelines/datasets/au_abs_cpi/flows.py        |  4 +-
 .../datasets/au_abs_labour_force/flows.py     |  4 +-
 .../br_anatel_banda_larga_fixa/flows.py       |  4 +-
 .../br_anatel_telefonia_movel/flows.py        |  4 +-
 .../br_anp_precos_combustiveis/flows.py       |  4 +-
 .../datasets/br_ans_beneficiario/flows.py     |  4 +-
 pipelines/datasets/br_bcb_agencia/flows.py    |  4 +-
 pipelines/datasets/br_bcb_estban/flows.py     |  4 +-
 pipelines/datasets/br_bcb_sicor/flows.py      |  4 +-
 .../datasets/br_bcb_taxa_cambio/flows.py      |  4 +-
 pipelines/datasets/br_bcb_taxa_selic/flows.py |  3 +-
 .../br_bndes_operacoes_contratadas/flows.py   |  8 ++--
 .../datasets/br_camara_dados_abertos/flows.py |  4 +-
 .../br_cgu_beneficios_cidadao/flows.py        |  4 +-
 .../datasets/br_cgu_cartao_pagamento/flows.py |  4 +-
 .../br_cgu_emendas_parlamentares/flows.py     |  4 +-
 .../br_cgu_licitacao_contrato/flows.py        |  6 +--
 .../br_cgu_pessoal_executivo_federal/flows.py |  4 +-
 .../flows.py                                  |  4 +-
 .../flows.py                                  |  4 +-
 .../br_cvm_administradores_carteira/flows.py  |  4 +-
 pipelines/datasets/br_cvm_fi/flows.py         |  4 +-
 .../flows.py                                  |  4 +-
 pipelines/datasets/br_denatran_frota/flows.py |  6 ++-
 pipelines/datasets/br_fgv_igp/flows.py        |  4 +-
 pipelines/datasets/br_ibge_inpc/flows.py      |  4 +-
 pipelines/datasets/br_ibge_ipca/flows.py      |  4 +-
 pipelines/datasets/br_ibge_ipca15/flows.py    |  4 +-
 pipelines/datasets/br_ibge_pnadc/flows.py     |  6 ++-
 pipelines/datasets/br_inmet_bdmep/flows.py    |  4 +-
 pipelines/datasets/br_me_caged/flows.py       |  4 +-
 pipelines/datasets/br_me_cnpj/flows.py        |  4 +-
 pipelines/datasets/br_me_comex_stat/flows.py  |  4 +-
 pipelines/datasets/br_me_siconfi/flows.py     |  4 +-
 pipelines/datasets/br_mp_pep/flows.py         |  4 +-
 pipelines/datasets/br_ms_cnes/flows.py        |  6 +--
 pipelines/datasets/br_ms_sia/flows.py         |  4 +-
 pipelines/datasets/br_ms_sih/flows.py         |  4 +-
 .../datasets/br_poder360_pesquisas/flows.py   |  4 +-
 pipelines/datasets/br_rf_cafir/flows.py       |  4 +-
 pipelines/datasets/br_rf_cno/flows.py         |  4 +-
 .../br_rj_isp_estatisticas_seguranca/flows.py |  4 +-
 .../datasets/br_senado_dados_abertos/flows.py |  4 +-
 pipelines/datasets/br_sfb_sicar/flows.py      |  4 +-
 .../datasets/br_stf_corte_aberta/flows.py     |  4 +-
 pipelines/datasets/br_tse_eleicoes/flows.py   |  4 +-
 pipelines/datasets/fundacao_lemann/flows.py   |  4 +-
 pipelines/datasets/us_bls_cpi/flows.py        |  4 +-
 pipelines/datasets/us_bls_qcew/flows.py       |  4 +-
 pipelines/datasets/world_cricsheet/flows.py   |  4 +-
 pipelines/utils/flow.py                       | 39 ++++++-------------
 pipelines/utils/tests/test_flow.py            |  9 +++--
 .../{{cookiecutter.pipeline_name}}/flows.py   |  6 ++-
 56 files changed, 184 insertions(+), 109 deletions(-)

diff --git a/.claude/rules/prefect-pipeline-conventions.md b/.claude/rules/prefect-pipeline-conventions.md
index 02b7230533..d8067f69bb 100644
--- a/.claude/rules/prefect-pipeline-conventions.md
+++ b/.claude/rules/prefect-pipeline-conventions.md
@@ -44,7 +44,8 @@ plain Prefect flow is a Pyrefly `missing-attribute` error. `pipelines/utils/flow
 declares both on a `prefect.Flow` subclass and exports a `flow` decorator that
 builds it; it takes the same arguments as `prefect.flow` and the object stays a
 `prefect.Flow` for every `isinstance` check. Both attributes default to empty
-(no schedule, work-pool default infrastructure).
+(no schedule, work-pool default infrastructure). `deploy_schedules` is a list of
+`prefect.schedules.Cron` — `Cron("0 16 10 * *", timezone="America/Sao_Paulo")`.
 
 ## DRY with the onboarding code
 
@@ -289,6 +290,8 @@ on the **deployed Prefect worker** (its pod SA has access) — the local
 Schedule inline on the flow object (do NOT register storage/run-config by hand):
 
 ```python
+from prefect.schedules import Cron
+
 from pipelines.utils.flow import flow
 
 
@@ -297,7 +300,7 @@ def my_flow() -> None: ...
 
 
 my_flow.deploy_schedules = [
-    {"cron": "0 16 10,11,12,13 * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 16 10,11,12,13 * *", timezone="America/Sao_Paulo")
 ]
 my_flow.job_variables = {
     "memory": "8Gi"
@@ -311,7 +314,7 @@ Deploy is CI, via `.github/scripts/deploy_flows.py`:
   looks fine. The workflow triggers on `labeled` and `synchronize`, so adding the
   label is itself enough. Schedules are **stripped** — manual runs only.
 - **Prod pool** (`cd-prefect3.yaml`, `--pool basedosdados --all`, on merge to main):
-  schedules become `Cron` objects; deployed **`paused=True`**.
+  `deploy_schedules` is passed straight to the deployment; deployed **`paused=True`**.
 - Cron in `America/Sao_Paulo`; see crontab.guru. For a monthly source, poll across
   a few release-window days — the source-poll guard no-ops until a new period lands.
 
diff --git a/.github/scripts/deploy_flows.py b/.github/scripts/deploy_flows.py
index d685ae5267..395bb7d65a 100644
--- a/.github/scripts/deploy_flows.py
+++ b/.github/scripts/deploy_flows.py
@@ -17,7 +17,6 @@
 
 from prefect import Flow
 from prefect.runner.storage import GitRepository
-from prefect.schedules import Cron
 
 REPO_URL = "https://github.com/basedosdados/pipelines.git"
 
@@ -75,14 +74,6 @@ def deploy_flow(
     schedules = getattr(flow, "deploy_schedules", None) or None
     if is_dev:
         schedules = None  # flows em dev não têm schedule
-    elif schedules:
-        # Convert dict {"cron": "...", "timezone": "..."} to Cron schedule objects
-        schedules = [
-            Cron(s["cron"], timezone=s.get("timezone", "UTC"))
-            if isinstance(s, dict)
-            else s
-            for s in schedules
-        ]
 
     job_variables = getattr(flow, "job_variables", None) or None
 
diff --git a/AGENTS.md b/AGENTS.md
index e1cae5cd69..395114b1ce 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -69,10 +69,12 @@ uv run manage.py add-pipeline <dataset_id>
 - `constants.py`: Use a `constants` enum or plain constants — no hardcoded values elsewhere.
 - `utils.py`: Pure helper functions with no Prefect decorators.
 
-There is no `schedules.py`. Attach the schedule to the flow object in `flows.py`; CI turns
-these dicts into `Cron` objects at deploy time:
+There is no `schedules.py`. Attach the schedule to the flow object in `flows.py`, as
+`Cron` objects from `prefect.schedules` (the `timezone` is an argument of `Cron`):
 
 ```python
+from prefect.schedules import Cron
+
 from pipelines.utils.flow import flow
 
 
@@ -80,9 +82,7 @@ from pipelines.utils.flow import flow
 def my_flow() -> None: ...
 
 
-my_flow.deploy_schedules = [
-    {"cron": "0 16 10 * *", "timezone": "America/Sao_Paulo"}
-]
+my_flow.deploy_schedules = [Cron("0 16 10 * *", timezone="America/Sao_Paulo")]
 my_flow.job_variables = {
     "memory": "8Gi"
 }  # optional; size to the flow's peak RAM
diff --git a/pipelines/datasets/au_abs_cpi/flows.py b/pipelines/datasets/au_abs_cpi/flows.py
index d39eabbde1..8d137f3296 100644
--- a/pipelines/datasets/au_abs_cpi/flows.py
+++ b/pipelines/datasets/au_abs_cpi/flows.py
@@ -15,6 +15,8 @@
 import shutil
 import tempfile
 
+from prefect.schedules import Cron
+
 from pipelines.datasets.au_abs_cpi.constants import constants
 from pipelines.datasets.au_abs_cpi.tasks import clean_cpi, download_cpi
 from pipelines.utils.flow import flow
@@ -173,5 +175,5 @@ def au_abs_cpi_flow(
 # 4th Wednesday from Feb 2027). Poll across the last week at 16:00 BRT; the
 # source-poll guard no-ops until a new month lands.
 au_abs_cpi_flow.deploy_schedules = [
-    {"cron": "0 16 22,23,24,25,26,27,28 * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 16 22,23,24,25,26,27,28 * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/au_abs_labour_force/flows.py b/pipelines/datasets/au_abs_labour_force/flows.py
index 9cd1858946..144d8d5b35 100644
--- a/pipelines/datasets/au_abs_labour_force/flows.py
+++ b/pipelines/datasets/au_abs_labour_force/flows.py
@@ -19,6 +19,8 @@
 import shutil
 import tempfile
 
+from prefect.schedules import Cron
+
 from pipelines.datasets.au_abs_labour_force.constants import constants
 from pipelines.datasets.au_abs_labour_force.tasks import (
     clean_and_write_task,
@@ -191,7 +193,7 @@ def au_abs_labour_force_flow(
 # 11:30 Canberra time. Poll daily across that window at 06:00 BRT (= evening AEST,
 # after the morning release); the source-poll guard no-ops until a new month lands.
 au_abs_labour_force_flow.deploy_schedules = [
-    {"cron": "0 6 14-27 * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 6 14-27 * *", timezone="America/Sao_Paulo")
 ]
 # openpyxl reads the ~38 MB SEM1 pivot; give the worker headroom.
 au_abs_labour_force_flow.job_variables = {"memory": "6Gi"}
diff --git a/pipelines/datasets/br_anatel_banda_larga_fixa/flows.py b/pipelines/datasets/br_anatel_banda_larga_fixa/flows.py
index a8e20d40a7..8138213c4b 100644
--- a/pipelines/datasets/br_anatel_banda_larga_fixa/flows.py
+++ b/pipelines/datasets/br_anatel_banda_larga_fixa/flows.py
@@ -5,6 +5,8 @@
 deste diretório.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.anatel.banda_larga_fixa.flows import (
     _run_anatel_banda_larga_fixa,
 )
@@ -39,7 +41,7 @@ def _flow(
         )
 
     _flow.deploy_schedules = (
-        [{"cron": cron, "timezone": "America/Sao_Paulo"}] if cron else []
+        [Cron(cron, timezone="America/Sao_Paulo")] if cron else []
     )
     return _flow
 
diff --git a/pipelines/datasets/br_anatel_telefonia_movel/flows.py b/pipelines/datasets/br_anatel_telefonia_movel/flows.py
index fcfda86327..d0750206f5 100644
--- a/pipelines/datasets/br_anatel_telefonia_movel/flows.py
+++ b/pipelines/datasets/br_anatel_telefonia_movel/flows.py
@@ -1,5 +1,7 @@
 """Flows for br_anatel_telefonia_movel — Prefect 3."""
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.anatel.telefonia_movel.flows import (
     _run_anatel_telefonia_movel,
 )
@@ -34,7 +36,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     _flow.job_variables = {"memory_limit": "8Gi", "memory_request": "2Gi"}
     return _flow
 
diff --git a/pipelines/datasets/br_anp_precos_combustiveis/flows.py b/pipelines/datasets/br_anp_precos_combustiveis/flows.py
index bed3ed9703..35d227526b 100644
--- a/pipelines/datasets/br_anp_precos_combustiveis/flows.py
+++ b/pipelines/datasets/br_anp_precos_combustiveis/flows.py
@@ -2,6 +2,8 @@
 Flow br_anp_precos_combustiveis__microdados — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.anp_precos_combustiveis.tasks import (
     download_and_transform,
     get_data_source_anp_max_date,
@@ -120,5 +122,5 @@ def br_anp_precos_combustiveis__microdados(
 
 
 br_anp_precos_combustiveis__microdados.deploy_schedules = [
-    {"cron": "0 10 * * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 10 * * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_ans_beneficiario/flows.py b/pipelines/datasets/br_ans_beneficiario/flows.py
index 608578a9a8..79ef9571f4 100644
--- a/pipelines/datasets/br_ans_beneficiario/flows.py
+++ b/pipelines/datasets/br_ans_beneficiario/flows.py
@@ -2,6 +2,8 @@
 Flow br_ans_beneficiario__informacao_consolidada — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.ans_beneficiario.tasks import (
     crawler_ans,
     extract_links_and_dates,
@@ -133,5 +135,5 @@ def br_ans_beneficiario__informacao_consolidada(
 
 
 br_ans_beneficiario__informacao_consolidada.deploy_schedules = [
-    {"cron": "0 21 * * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 21 * * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_bcb_agencia/flows.py b/pipelines/datasets/br_bcb_agencia/flows.py
index 05c4edf759..4dd66c2130 100644
--- a/pipelines/datasets/br_bcb_agencia/flows.py
+++ b/pipelines/datasets/br_bcb_agencia/flows.py
@@ -2,6 +2,8 @@
 Flow br_bcb_agencia__agencia — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.bcb_agencia.tasks import (
     clean_data,
     download_table,
@@ -146,5 +148,5 @@ def br_bcb_agencia__agencia(
 
 
 br_bcb_agencia__agencia.deploy_schedules = [
-    {"cron": "0 22 25-31 * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 22 25-31 * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_bcb_estban/flows.py b/pipelines/datasets/br_bcb_estban/flows.py
index 19a80e85fb..4b047307cb 100644
--- a/pipelines/datasets/br_bcb_estban/flows.py
+++ b/pipelines/datasets/br_bcb_estban/flows.py
@@ -2,6 +2,8 @@
 Flows para br_bcb_estban — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.bcb_estban.tasks import (
     cleaning_data,
     download_table,
@@ -166,7 +168,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_bcb_sicor/flows.py b/pipelines/datasets/br_bcb_sicor/flows.py
index 4570c672a4..219ff15661 100644
--- a/pipelines/datasets/br_bcb_sicor/flows.py
+++ b/pipelines/datasets/br_bcb_sicor/flows.py
@@ -2,6 +2,8 @@
 Flows para br_bcb_sicor — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.bcb.flows import _run_bcb_sicor
 from pipelines.crawler.bcb.tasks import create_load_dictionary
 from pipelines.utils.flow import flow
@@ -51,7 +53,7 @@ def _flow(
             local_redis_execution=local_redis_execution,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_bcb_taxa_cambio/flows.py b/pipelines/datasets/br_bcb_taxa_cambio/flows.py
index 42b2c47296..8834410576 100644
--- a/pipelines/datasets/br_bcb_taxa_cambio/flows.py
+++ b/pipelines/datasets/br_bcb_taxa_cambio/flows.py
@@ -2,6 +2,8 @@
 Flow br_bcb_taxa_cambio — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.bcb_taxa_cambio.tasks import (
     get_data_taxa_cambio,
     treat_data_taxa_cambio,
@@ -94,5 +96,5 @@ def br_bcb_taxa_cambio__taxa_cambio(
 
 
 br_bcb_taxa_cambio__taxa_cambio.deploy_schedules = [
-    {"cron": "0 8 * * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 8 * * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_bcb_taxa_selic/flows.py b/pipelines/datasets/br_bcb_taxa_selic/flows.py
index 1750a7b208..3eedc0a440 100644
--- a/pipelines/datasets/br_bcb_taxa_selic/flows.py
+++ b/pipelines/datasets/br_bcb_taxa_selic/flows.py
@@ -7,6 +7,7 @@
 import pandas as pd
 import requests
 from prefect import task
+from prefect.schedules import Cron
 
 from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
@@ -156,5 +157,5 @@ def br_bcb_taxa_selic__taxa_selic(
 
 
 br_bcb_taxa_selic__taxa_selic.deploy_schedules = [
-    {"cron": "0 8 * * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 8 * * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_bndes_operacoes_contratadas/flows.py b/pipelines/datasets/br_bndes_operacoes_contratadas/flows.py
index eee6d1658a..cc7f71f3e0 100644
--- a/pipelines/datasets/br_bndes_operacoes_contratadas/flows.py
+++ b/pipelines/datasets/br_bndes_operacoes_contratadas/flows.py
@@ -5,6 +5,8 @@
 orquestracao (poll deferido) vive em pipelines/crawler/bndes/flows.py.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.bndes.flows import (
     _run_operacoes,
     _run_operacoes_administracao_publica,
@@ -41,7 +43,7 @@ def br_bndes_operacoes_contratadas__operacoes_indiretas_automaticas(
 
 
 br_bndes_operacoes_contratadas__operacoes_indiretas_automaticas.deploy_schedules = [
-    {"cron": "0 6 * * 1", "timezone": "America/Sao_Paulo"}
+    Cron("0 6 * * 1", timezone="America/Sao_Paulo")
 ]
 
 
@@ -74,7 +76,7 @@ def br_bndes_operacoes_contratadas__operacoes_nao_automaticas(
 
 
 br_bndes_operacoes_contratadas__operacoes_nao_automaticas.deploy_schedules = [
-    {"cron": "0 6 * * 1", "timezone": "America/Sao_Paulo"}
+    Cron("0 6 * * 1", timezone="America/Sao_Paulo")
 ]
 
 
@@ -109,5 +111,5 @@ def br_bndes_operacoes_contratadas__operacoes_administracao_publica(
 # cron semanal (segunda 06h BRT), igual a irma; a fonte atualiza mensal e o poll
 # deferido no-opa quando nao ha novidade. Ajuste se quiser outra janela.
 br_bndes_operacoes_contratadas__operacoes_administracao_publica.deploy_schedules = [
-    {"cron": "0 6 * * 1", "timezone": "America/Sao_Paulo"}
+    Cron("0 6 * * 1", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_camara_dados_abertos/flows.py b/pipelines/datasets/br_camara_dados_abertos/flows.py
index 0e63a0ad86..7d88ca079b 100644
--- a/pipelines/datasets/br_camara_dados_abertos/flows.py
+++ b/pipelines/datasets/br_camara_dados_abertos/flows.py
@@ -2,6 +2,8 @@
 Flows para br_camara_dados_abertos — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.camara_dados_abertos.flows import (
     _run_camara_dados_abertos,
 )
@@ -32,7 +34,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_cgu_beneficios_cidadao/flows.py b/pipelines/datasets/br_cgu_beneficios_cidadao/flows.py
index 78e616a3b3..f5113774fc 100644
--- a/pipelines/datasets/br_cgu_beneficios_cidadao/flows.py
+++ b/pipelines/datasets/br_cgu_beneficios_cidadao/flows.py
@@ -6,6 +6,8 @@
 em pipelines/crawler/cgu/utils.py e pipelines/utils/utils.py.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.cgu.flows import _run_cgu_beneficios_cidadao
 from pipelines.utils.flow import flow
 
@@ -36,7 +38,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_cgu_cartao_pagamento/flows.py b/pipelines/datasets/br_cgu_cartao_pagamento/flows.py
index 831230e0bf..76924317cb 100644
--- a/pipelines/datasets/br_cgu_cartao_pagamento/flows.py
+++ b/pipelines/datasets/br_cgu_cartao_pagamento/flows.py
@@ -2,6 +2,8 @@
 Flows para br_cgu_cartao_pagamento — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.cgu.flows import _run_cgu_cartao_pagamento
 from pipelines.utils.flow import flow
 
@@ -32,7 +34,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_cgu_emendas_parlamentares/flows.py b/pipelines/datasets/br_cgu_emendas_parlamentares/flows.py
index 47b5c9c915..2529a4d9e0 100644
--- a/pipelines/datasets/br_cgu_emendas_parlamentares/flows.py
+++ b/pipelines/datasets/br_cgu_emendas_parlamentares/flows.py
@@ -2,6 +2,8 @@
 Flows para br_cgu_emendas_parlamentares — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.cgu_emendas_parlamentares.tasks import (
     convert_str_to_float,
     get_last_modified_time,
@@ -119,5 +121,5 @@ def br_cgu_emendas_parlamentares__microdados(
 
 
 br_cgu_emendas_parlamentares__microdados.deploy_schedules = [
-    {"cron": "30 19 * * *", "timezone": "America/Sao_Paulo"}
+    Cron("30 19 * * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_cgu_licitacao_contrato/flows.py b/pipelines/datasets/br_cgu_licitacao_contrato/flows.py
index 5538722191..05fad06df1 100644
--- a/pipelines/datasets/br_cgu_licitacao_contrato/flows.py
+++ b/pipelines/datasets/br_cgu_licitacao_contrato/flows.py
@@ -2,6 +2,8 @@
 Flows para br_cgu_licitacao_contrato — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.cgu.flows import _run_cgu_licitacao_contrato
 from pipelines.utils.flow import flow
 
@@ -33,9 +35,7 @@ def _flow(
         )
 
     if cron:
-        _flow.deploy_schedules = [
-            {"cron": cron, "timezone": "America/Sao_Paulo"}
-        ]
+        _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_cgu_pessoal_executivo_federal/flows.py b/pipelines/datasets/br_cgu_pessoal_executivo_federal/flows.py
index e09d0809f0..a106e1dcc8 100644
--- a/pipelines/datasets/br_cgu_pessoal_executivo_federal/flows.py
+++ b/pipelines/datasets/br_cgu_pessoal_executivo_federal/flows.py
@@ -1,5 +1,7 @@
 """Flows para br_cgu_pessoal_executivo_federal — Prefect 3."""
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.cgu_pessoal_executivo_federal.tasks import (
     clean_save_table,
     crawl,
@@ -75,5 +77,5 @@ def br_cgu_pessoal_executivo_federal__terceirizados(
 
 
 br_cgu_pessoal_executivo_federal__terceirizados.deploy_schedules = [
-    {"cron": "0 0 28 2/4 *", "timezone": "America/Sao_Paulo"}
+    Cron("0 0 28 2/4 *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_cgu_servidores_executivo_federal/flows.py b/pipelines/datasets/br_cgu_servidores_executivo_federal/flows.py
index 71ec79abe6..bfc9939874 100644
--- a/pipelines/datasets/br_cgu_servidores_executivo_federal/flows.py
+++ b/pipelines/datasets/br_cgu_servidores_executivo_federal/flows.py
@@ -17,6 +17,8 @@
 enquanto o ZIP é gerado de forma assíncrona).
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.cgu.flows import _run_cgu_servidores_publicos
 from pipelines.utils.flow import flow
 
@@ -47,7 +49,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_cnj_improbidade_administrativa/flows.py b/pipelines/datasets/br_cnj_improbidade_administrativa/flows.py
index 4b9db6707f..da0aca3369 100644
--- a/pipelines/datasets/br_cnj_improbidade_administrativa/flows.py
+++ b/pipelines/datasets/br_cnj_improbidade_administrativa/flows.py
@@ -2,6 +2,8 @@
 Flow br_cnj_improbidade_administrativa — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.cnj_improbidade_administrativa.tasks import (
     get_max_date,
     is_up_to_date,
@@ -99,5 +101,5 @@ def br_cnj_improbidade_administrativa__condenacao(
 
 
 br_cnj_improbidade_administrativa__condenacao.deploy_schedules = [
-    {"cron": "0 7 * * 1", "timezone": "America/Sao_Paulo"}
+    Cron("0 7 * * 1", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_cvm_administradores_carteira/flows.py b/pipelines/datasets/br_cvm_administradores_carteira/flows.py
index fa199e2c78..7ea108d12d 100644
--- a/pipelines/datasets/br_cvm_administradores_carteira/flows.py
+++ b/pipelines/datasets/br_cvm_administradores_carteira/flows.py
@@ -2,6 +2,8 @@
 Flows for br_cvm_administradores_carteira — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.cvm_administradores_carteira.flows import (
     _run_cvm_administradores_carteira,
 )
@@ -32,7 +34,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_cvm_fi/flows.py b/pipelines/datasets/br_cvm_fi/flows.py
index 30d7a84e1e..e169dc7e34 100644
--- a/pipelines/datasets/br_cvm_fi/flows.py
+++ b/pipelines/datasets/br_cvm_fi/flows.py
@@ -2,6 +2,8 @@
 Flows for br_cvm_fi — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.cvm.flows import _run_cvm_fi
 from pipelines.utils.flow import flow
 
@@ -33,7 +35,7 @@ def _flow(
             url=url,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_cvm_oferta_publica_distribuicao/flows.py b/pipelines/datasets/br_cvm_oferta_publica_distribuicao/flows.py
index e8a4439963..1961da5360 100644
--- a/pipelines/datasets/br_cvm_oferta_publica_distribuicao/flows.py
+++ b/pipelines/datasets/br_cvm_oferta_publica_distribuicao/flows.py
@@ -2,6 +2,8 @@
 Flows for br_cvm_oferta_publica_distribuicao — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.cvm_oferta_publica_distribuicao.tasks import (
     clean_table_oferta_distribuicao,
     crawl,
@@ -95,5 +97,5 @@ def br_cvm_oferta_publica_distribuicao__dia(
 
 
 br_cvm_oferta_publica_distribuicao__dia.deploy_schedules = [
-    {"cron": "45 6 * * 1-5", "timezone": "America/Sao_Paulo"}
+    Cron("45 6 * * 1-5", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_denatran_frota/flows.py b/pipelines/datasets/br_denatran_frota/flows.py
index f8e49ca1bd..7ad0f2ca91 100644
--- a/pipelines/datasets/br_denatran_frota/flows.py
+++ b/pipelines/datasets/br_denatran_frota/flows.py
@@ -2,6 +2,8 @@
 Flows para br_denatran_frota — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.denatran_frota.constants import (
     constants as denatran_constants,
 )
@@ -197,8 +199,8 @@ def br_denatran_frota__municipio_tipo(
 
 
 br_denatran_frota__uf_tipo.deploy_schedules = [
-    {"cron": "0 21 10-30 * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 21 10-30 * *", timezone="America/Sao_Paulo")
 ]
 br_denatran_frota__municipio_tipo.deploy_schedules = [
-    {"cron": "20 21 10-30 * *", "timezone": "America/Sao_Paulo"}
+    Cron("20 21 10-30 * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_fgv_igp/flows.py b/pipelines/datasets/br_fgv_igp/flows.py
index f22776cf4c..876626f4f4 100644
--- a/pipelines/datasets/br_fgv_igp/flows.py
+++ b/pipelines/datasets/br_fgv_igp/flows.py
@@ -2,6 +2,8 @@
 Flows for br_fgv_igp — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.fgv_igp.flows import _run_fgv_igp
 from pipelines.utils.flow import flow
 
@@ -33,7 +35,7 @@ def _flow(
         )
 
     _flow.deploy_schedules = (
-        [{"cron": cron, "timezone": "America/Sao_Paulo"}] if cron else []
+        [Cron(cron, timezone="America/Sao_Paulo")] if cron else []
     )
     return _flow
 
diff --git a/pipelines/datasets/br_ibge_inpc/flows.py b/pipelines/datasets/br_ibge_inpc/flows.py
index c501a1f119..a388da1f80 100644
--- a/pipelines/datasets/br_ibge_inpc/flows.py
+++ b/pipelines/datasets/br_ibge_inpc/flows.py
@@ -2,6 +2,8 @@
 Flows para br_ibge_inpc — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.ibge_inflacao.flows import _run_ibge_inflacao
 from pipelines.utils.flow import flow
 
@@ -29,7 +31,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_ibge_ipca/flows.py b/pipelines/datasets/br_ibge_ipca/flows.py
index a1d62a2d44..5dc9d050a2 100644
--- a/pipelines/datasets/br_ibge_ipca/flows.py
+++ b/pipelines/datasets/br_ibge_ipca/flows.py
@@ -5,6 +5,8 @@
 (guards contra bloco vazio da API do IBGE) em utils.py.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.ibge_inflacao.flows import _run_ibge_inflacao
 from pipelines.utils.flow import flow
 
@@ -32,7 +34,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_ibge_ipca15/flows.py b/pipelines/datasets/br_ibge_ipca15/flows.py
index 997bd1b6a9..ecc22f3413 100644
--- a/pipelines/datasets/br_ibge_ipca15/flows.py
+++ b/pipelines/datasets/br_ibge_ipca15/flows.py
@@ -2,6 +2,8 @@
 Flows para br_ibge_ipca15 — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.ibge_inflacao.flows import _run_ibge_inflacao
 from pipelines.utils.flow import flow
 
@@ -32,7 +34,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_ibge_pnadc/flows.py b/pipelines/datasets/br_ibge_pnadc/flows.py
index f975915ee2..a08435faee 100644
--- a/pipelines/datasets/br_ibge_pnadc/flows.py
+++ b/pipelines/datasets/br_ibge_pnadc/flows.py
@@ -2,6 +2,8 @@
 Flow br_ibge_pnadc — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.ibge_pnadc.tasks import (
     build_partitions,
     build_table_paths,
@@ -121,7 +123,7 @@ def br_ibge_pnadc__microdados(
 
 
 br_ibge_pnadc__microdados.deploy_schedules = [
-    {"cron": "0 5 15-31 2,5,8,11 *", "timezone": "America/Sao_Paulo"}
+    Cron("0 5 15-31 2,5,8,11 *", timezone="America/Sao_Paulo")
 ]
 
 
@@ -195,5 +197,5 @@ def br_ibge_pnadc__dicionario(
 
 
 br_ibge_pnadc__dicionario.deploy_schedules = [
-    {"cron": "0 5 1,15 * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 5 1,15 * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_inmet_bdmep/flows.py b/pipelines/datasets/br_inmet_bdmep/flows.py
index 0ca882c000..ad1fd2f754 100644
--- a/pipelines/datasets/br_inmet_bdmep/flows.py
+++ b/pipelines/datasets/br_inmet_bdmep/flows.py
@@ -2,6 +2,8 @@
 Flows for br_inmet_bdmep — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.inmet_bdmep.tasks import (
     extract_last_date_from_source,
     get_base_inmet,
@@ -115,5 +117,5 @@ def br_inmet_bdmep__microdados(
 
 
 br_inmet_bdmep__microdados.deploy_schedules = [
-    {"cron": "0 22 * * 1-5", "timezone": "America/Sao_Paulo"},
+    Cron("0 22 * * 1-5", timezone="America/Sao_Paulo"),
 ]
diff --git a/pipelines/datasets/br_me_caged/flows.py b/pipelines/datasets/br_me_caged/flows.py
index 588bbc3f87..ff07cd9cba 100644
--- a/pipelines/datasets/br_me_caged/flows.py
+++ b/pipelines/datasets/br_me_caged/flows.py
@@ -2,6 +2,8 @@
 Flows para br_me_caged — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.me_caged.tasks import (
     build_partitions,
     build_table_paths,
@@ -148,7 +150,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_me_cnpj/flows.py b/pipelines/datasets/br_me_cnpj/flows.py
index a77ded7d6a..80f0fb3e6c 100644
--- a/pipelines/datasets/br_me_cnpj/flows.py
+++ b/pipelines/datasets/br_me_cnpj/flows.py
@@ -2,6 +2,8 @@
 Flows for br_me_cnpj — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.me_cnpj.flows import _run_me_cnpj
 from pipelines.utils.flow import flow
 
@@ -30,7 +32,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_me_comex_stat/flows.py b/pipelines/datasets/br_me_comex_stat/flows.py
index b7a05a975f..ce198933a5 100644
--- a/pipelines/datasets/br_me_comex_stat/flows.py
+++ b/pipelines/datasets/br_me_comex_stat/flows.py
@@ -2,6 +2,8 @@
 Flows for br_me_comex_stat — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.me_comex_stat.constants import (
     constants as comex_constants,
 )
@@ -129,7 +131,7 @@ def _flow(
                     date_format="%Y-%m",
                 )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_me_siconfi/flows.py b/pipelines/datasets/br_me_siconfi/flows.py
index 822ec312d8..b17a3252e5 100644
--- a/pipelines/datasets/br_me_siconfi/flows.py
+++ b/pipelines/datasets/br_me_siconfi/flows.py
@@ -23,6 +23,8 @@
 import tempfile
 from datetime import datetime
 
+from prefect.schedules import Cron
+
 from pipelines.datasets.br_me_siconfi import tasks, utils
 from pipelines.datasets.br_me_siconfi.constants import constants
 from pipelines.utils.flow import flow
@@ -213,7 +215,7 @@ def br_me_siconfi_flow(
 # SICONFI is annual but revised retroactively; rebuild once a month (1st at
 # 16:00 BRT). Each run rebuilds fully — there is no source-poll no-op here.
 br_me_siconfi_flow.deploy_schedules = [
-    {"cron": "0 16 1 * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 16 1 * *", timezone="America/Sao_Paulo")
 ]
 # The município window build holds a full year of data in pandas at a time.
 br_me_siconfi_flow.job_variables = {"memory": "16Gi"}
diff --git a/pipelines/datasets/br_mp_pep/flows.py b/pipelines/datasets/br_mp_pep/flows.py
index cae2cebdcc..d46148e60d 100644
--- a/pipelines/datasets/br_mp_pep/flows.py
+++ b/pipelines/datasets/br_mp_pep/flows.py
@@ -4,6 +4,8 @@
 
 import datetime
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.mp_pep.tasks import (
     clean_data,
     download_xlsx,
@@ -112,5 +114,5 @@ def br_mp_pep__cargos_funcoes(
 
 
 br_mp_pep__cargos_funcoes.deploy_schedules = [
-    {"cron": "0 14 * * 3", "timezone": "America/Sao_Paulo"}
+    Cron("0 14 * * 3", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_ms_cnes/flows.py b/pipelines/datasets/br_ms_cnes/flows.py
index f8acd40c09..90828086d9 100644
--- a/pipelines/datasets/br_ms_cnes/flows.py
+++ b/pipelines/datasets/br_ms_cnes/flows.py
@@ -13,6 +13,8 @@
 este arquivo, o deploy sai "0 registrados, N pulados" e passa.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.datasus.flows import _run_cnes
 from pipelines.utils.flow import flow
 
@@ -44,9 +46,7 @@ def _flow(
         )
 
     if cron:
-        _flow.deploy_schedules = [
-            {"cron": cron, "timezone": "America/Sao_Paulo"}
-        ]
+        _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_ms_sia/flows.py b/pipelines/datasets/br_ms_sia/flows.py
index 30e7b0cc56..5ac0ed0b9e 100644
--- a/pipelines/datasets/br_ms_sia/flows.py
+++ b/pipelines/datasets/br_ms_sia/flows.py
@@ -2,6 +2,8 @@
 Flows for br_ms_sia — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.datasus.flows import _run_siasus
 from pipelines.utils.flow import flow
 
@@ -32,7 +34,7 @@ def _flow(
             year_month_to_extract=year_month_to_extract,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_ms_sih/flows.py b/pipelines/datasets/br_ms_sih/flows.py
index a21787f2f4..d6d0332956 100644
--- a/pipelines/datasets/br_ms_sih/flows.py
+++ b/pipelines/datasets/br_ms_sih/flows.py
@@ -2,6 +2,8 @@
 Flows for br_ms_sih — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.datasus.flows import _run_sihsus
 from pipelines.utils.flow import flow
 
@@ -32,7 +34,7 @@ def _flow(
             year_month_to_extract=year_month_to_extract,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_poder360_pesquisas/flows.py b/pipelines/datasets/br_poder360_pesquisas/flows.py
index 81e02b1c02..3b01c7ef24 100644
--- a/pipelines/datasets/br_poder360_pesquisas/flows.py
+++ b/pipelines/datasets/br_poder360_pesquisas/flows.py
@@ -2,6 +2,8 @@
 Flow br_poder360_pesquisas — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.poder360_pesquisas.tasks import crawler
 from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
@@ -88,5 +90,5 @@ def br_poder360_pesquisas__microdados(
 
 
 br_poder360_pesquisas__microdados.deploy_schedules = [
-    {"cron": "42 3 * * *", "timezone": "America/Sao_Paulo"}
+    Cron("42 3 * * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_rf_cafir/flows.py b/pipelines/datasets/br_rf_cafir/flows.py
index 6074c62d2d..c1813ffd6a 100644
--- a/pipelines/datasets/br_rf_cafir/flows.py
+++ b/pipelines/datasets/br_rf_cafir/flows.py
@@ -2,6 +2,8 @@
 Flows for br_rf_cafir — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.rf_cafir.constants import (
     constants as br_rf_cafir_constants,
 )
@@ -134,5 +136,5 @@ def br_rf_cafir__imoveis_rurais(
 
 
 br_rf_cafir__imoveis_rurais.deploy_schedules = [
-    {"cron": "0 0 * * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 0 * * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_rf_cno/flows.py b/pipelines/datasets/br_rf_cno/flows.py
index 17ab200cf9..ba3324a170 100644
--- a/pipelines/datasets/br_rf_cno/flows.py
+++ b/pipelines/datasets/br_rf_cno/flows.py
@@ -9,6 +9,8 @@
 `safe_cast(data as date)` do model virava NULL e o filtro incremental nunca inseria.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.rf.flows import _run_rf
 from pipelines.utils.flow import flow
 
@@ -39,7 +41,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_rj_isp_estatisticas_seguranca/flows.py b/pipelines/datasets/br_rj_isp_estatisticas_seguranca/flows.py
index d7fa71b929..50bf628782 100644
--- a/pipelines/datasets/br_rj_isp_estatisticas_seguranca/flows.py
+++ b/pipelines/datasets/br_rj_isp_estatisticas_seguranca/flows.py
@@ -2,6 +2,8 @@
 Flows for br_rj_isp_estatisticas_seguranca — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.isp.flows import _run_isp
 from pipelines.utils.flow import flow
 
@@ -30,7 +32,7 @@ def _flow(
             force_run=force_run,
         )
 
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 
diff --git a/pipelines/datasets/br_senado_dados_abertos/flows.py b/pipelines/datasets/br_senado_dados_abertos/flows.py
index dfad7b2c59..e140139a88 100644
--- a/pipelines/datasets/br_senado_dados_abertos/flows.py
+++ b/pipelines/datasets/br_senado_dados_abertos/flows.py
@@ -16,6 +16,8 @@
 import shutil
 import tempfile
 
+from prefect.schedules import Cron
+
 from pipelines.datasets.br_senado_dados_abertos.constants import constants
 from pipelines.datasets.br_senado_dados_abertos.tasks import extract_clean
 from pipelines.utils.flow import flow
@@ -165,6 +167,6 @@ def br_senado_dados_abertos_flow(
 
 # Legislative activity updates on business days; refresh every morning (BRT).
 br_senado_dados_abertos_flow.deploy_schedules = [
-    {"cron": "0 8 * * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 8 * * *", timezone="America/Sao_Paulo")
 ]
 br_senado_dados_abertos_flow.job_variables = {"memory": "4Gi"}
diff --git a/pipelines/datasets/br_sfb_sicar/flows.py b/pipelines/datasets/br_sfb_sicar/flows.py
index 4e2c8b8f93..16ab1a0046 100644
--- a/pipelines/datasets/br_sfb_sicar/flows.py
+++ b/pipelines/datasets/br_sfb_sicar/flows.py
@@ -2,6 +2,8 @@
 Flow br_sfb_sicar — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.sfb_sicar.constants import Constants
 from pipelines.crawler.sfb_sicar.tasks import (
     download_car,
@@ -113,5 +115,5 @@ def br_sfb_sicar__area_imovel(
 
 
 br_sfb_sicar__area_imovel.deploy_schedules = [
-    {"cron": "15 21 15 * *", "timezone": "America/Sao_Paulo"}
+    Cron("15 21 15 * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_stf_corte_aberta/flows.py b/pipelines/datasets/br_stf_corte_aberta/flows.py
index 0106ef149a..162c8df559 100644
--- a/pipelines/datasets/br_stf_corte_aberta/flows.py
+++ b/pipelines/datasets/br_stf_corte_aberta/flows.py
@@ -2,6 +2,8 @@
 Flow br_stf_corte_aberta — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.stf_corte_aberta.tasks import (
     download_and_transform,
     get_data_source_stf_max_date,
@@ -119,5 +121,5 @@ def br_stf_corte_aberta__decisoes(
 
 
 br_stf_corte_aberta__decisoes.deploy_schedules = [
-    {"cron": "0 12 * * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 12 * * *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/br_tse_eleicoes/flows.py b/pipelines/datasets/br_tse_eleicoes/flows.py
index a1a1815cdb..340e982ad3 100644
--- a/pipelines/datasets/br_tse_eleicoes/flows.py
+++ b/pipelines/datasets/br_tse_eleicoes/flows.py
@@ -2,6 +2,8 @@
 Flows for br_tse_eleicoes — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.crawler.tse_eleicoes.flows import _run_tse_eleicoes
 from pipelines.utils.flow import flow
 
@@ -31,7 +33,7 @@ def _flow(
         )
 
     _flow.deploy_schedules = (
-        [{"cron": cron, "timezone": "America/Sao_Paulo"}] if cron else []
+        [Cron(cron, timezone="America/Sao_Paulo")] if cron else []
     )
     return _flow
 
diff --git a/pipelines/datasets/fundacao_lemann/flows.py b/pipelines/datasets/fundacao_lemann/flows.py
index eac9baf956..f2b64ab202 100644
--- a/pipelines/datasets/fundacao_lemann/flows.py
+++ b/pipelines/datasets/fundacao_lemann/flows.py
@@ -2,6 +2,8 @@
 Flow fundacao_lemann — Prefect 3.
 """
 
+from prefect.schedules import Cron
+
 from pipelines.utils.flow import flow
 from pipelines.utils.tasks import (
     download_data_to_gcs,
@@ -40,5 +42,5 @@ def fundacao_lemann__ano_escola_serie_educacao_aprendizagem_adequada(
 
 
 fundacao_lemann__ano_escola_serie_educacao_aprendizagem_adequada.deploy_schedules = [
-    {"cron": "0 9 1 1 *", "timezone": "America/Sao_Paulo"}
+    Cron("0 9 1 1 *", timezone="America/Sao_Paulo")
 ]
diff --git a/pipelines/datasets/us_bls_cpi/flows.py b/pipelines/datasets/us_bls_cpi/flows.py
index 4e9b3721f1..437b5392e2 100644
--- a/pipelines/datasets/us_bls_cpi/flows.py
+++ b/pipelines/datasets/us_bls_cpi/flows.py
@@ -13,6 +13,8 @@
 import shutil
 import tempfile
 
+from prefect.schedules import Cron
+
 from pipelines.datasets.us_bls_cpi.constants import constants
 from pipelines.datasets.us_bls_cpi.tasks import clean_cpi, download_cpi
 from pipelines.utils.flow import flow
@@ -177,7 +179,7 @@ def us_bls_cpi_flow(
 # mid-month days at 16:00 BRT; the source-poll guard no-ops until a new month
 # actually appears.
 us_bls_cpi_flow.deploy_schedules = [
-    {"cron": "0 16 10,11,12,13,14,15 * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 16 10,11,12,13,14,15 * *", timezone="America/Sao_Paulo")
 ]
 # The clean step holds ~4M rows in pandas; give the worker headroom.
 us_bls_cpi_flow.job_variables = {"memory": "8Gi"}
diff --git a/pipelines/datasets/us_bls_qcew/flows.py b/pipelines/datasets/us_bls_qcew/flows.py
index d1cd883a01..b13224d306 100644
--- a/pipelines/datasets/us_bls_qcew/flows.py
+++ b/pipelines/datasets/us_bls_qcew/flows.py
@@ -19,6 +19,8 @@
 import shutil
 import tempfile
 
+from prefect.schedules import Cron
+
 from pipelines.datasets.us_bls_qcew.constants import constants
 from pipelines.datasets.us_bls_qcew.tasks import (
     clean_qcew,
@@ -188,7 +190,7 @@ def us_bls_qcew_flow(
 # Poll across the first ~10 days of those months at 16:00 BRT; the source-poll
 # guard no-ops until a new quarter actually appears in the singlefiles.
 us_bls_qcew_flow.deploy_schedules = [
-    {"cron": "0 16 1-10 3,6,9,12 *", "timezone": "America/Sao_Paulo"}
+    Cron("0 16 1-10 3,6,9,12 *", timezone="America/Sao_Paulo")
 ]
 # The clean step streams ~15M-row singlefiles one chunk at a time (peak ~1.75GB
 # in pandas); give the worker headroom above that.
diff --git a/pipelines/datasets/world_cricsheet/flows.py b/pipelines/datasets/world_cricsheet/flows.py
index bf0f09cdef..dffb41c83b 100644
--- a/pipelines/datasets/world_cricsheet/flows.py
+++ b/pipelines/datasets/world_cricsheet/flows.py
@@ -16,6 +16,8 @@
 import shutil
 import tempfile
 
+from prefect.schedules import Cron
+
 from pipelines.datasets.world_cricsheet.constants import constants
 from pipelines.datasets.world_cricsheet.tasks import (
     clean_cricsheet,
@@ -212,7 +214,7 @@ def world_cricsheet_flow(
 # freshness fine. The source-poll guard still no-ops between real releases, and
 # the full-replace dump means overlapping windows never duplicate.
 world_cricsheet_flow.deploy_schedules = [
-    {"cron": "0 6 * * 1", "timezone": "America/Sao_Paulo"}
+    Cron("0 6 * * 1", timezone="America/Sao_Paulo")
 ]
 # The deliveries build streams 11.4M rows and the bundle extracts to several GB;
 # give the worker headroom.
diff --git a/pipelines/utils/flow.py b/pipelines/utils/flow.py
index 33841e654a..c7f01e2dfe 100644
--- a/pipelines/utils/flow.py
+++ b/pipelines/utils/flow.py
@@ -4,9 +4,8 @@
 `.github/scripts/deploy_flows.py` lê dois atributos do objeto flow que o
 `prefect.Flow` não declara:
 
-- `deploy_schedules`: lista de agendamentos cron, convertida em objetos
-  `prefect.schedules.Cron` no deploy de produção (no pool de dev os schedules
-  são descartados);
+- `deploy_schedules`: lista de agendamentos (`prefect.schedules.Cron`), usada no
+  deploy de produção (no pool de dev os schedules são descartados);
 - `job_variables`: overrides da configuração de infraestrutura do work pool
   (memória, CPU...).
 
@@ -17,6 +16,8 @@
 `prefect.flow`:
 
 ```python
+from prefect.schedules import Cron
+
 from pipelines.utils.flow import flow
 
 
@@ -25,7 +26,7 @@ def meu_dataset_flow() -> None: ...
 
 
 meu_dataset_flow.deploy_schedules = [
-    {"cron": "0 16 10 * *", "timezone": "America/Sao_Paulo"}
+    Cron("0 16 10 * *", timezone="America/Sao_Paulo")
 ]
 meu_dataset_flow.job_variables = {"memory": "8Gi"}
 ```
@@ -35,47 +36,29 @@ def meu_dataset_flow() -> None: ...
 """
 
 from collections.abc import Callable
-from typing import (
-    Any,
-    NotRequired,
-    ParamSpec,
-    TypedDict,
-    TypeVar,
-    overload,
-)
+from typing import Any, ParamSpec, TypeVar, overload
 
 from prefect import Flow as PrefectFlow
 from prefect.futures import PrefectFuture
+from prefect.schedules import Schedule
 from prefect.task_runners import TaskRunner
 
 P = ParamSpec("P")
 R = TypeVar("R")
 
 
-class DeploySchedule(TypedDict):
-    """Agendamento de um flow, no formato lido por `deploy_flows.py`.
-
-    Attributes:
-        cron: Expressão cron (ver crontab.guru).
-        timezone: Fuso da expressão; `"UTC"` quando omitido. Por convenção do
-            repositório, use `"America/Sao_Paulo"`.
-    """
-
-    cron: str
-    timezone: NotRequired[str]
-
-
 class Flow(PrefectFlow[P, R]):
     """Flow do Prefect 3 com os atributos que o deploy da BD lê.
 
     Attributes:
-        deploy_schedules: Agendamentos do flow. Vazio (o padrão) significa
-            deployment sem schedule — execução apenas manual.
+        deploy_schedules: Agendamentos do flow, construídos com
+            `prefect.schedules.Cron` (que já recebe `timezone`). Vazio (o
+            padrão) significa deployment sem schedule — execução apenas manual.
         job_variables: Overrides da configuração de infraestrutura do work
             pool, por exemplo `{"memory": "8Gi"}`. Vazio usa o padrão do pool.
     """
 
-    deploy_schedules: list[DeploySchedule]
+    deploy_schedules: list[Schedule]
     job_variables: dict[str, Any]
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
diff --git a/pipelines/utils/tests/test_flow.py b/pipelines/utils/tests/test_flow.py
index 8ac628433c..9c109ba503 100644
--- a/pipelines/utils/tests/test_flow.py
+++ b/pipelines/utils/tests/test_flow.py
@@ -1,8 +1,9 @@
 """Testes do decorator `flow` da BD (`pipelines.utils.flow`)."""
 
 from prefect import Flow as PrefectFlow
+from prefect.schedules import Cron
 
-from pipelines.utils.flow import DeploySchedule, Flow, flow
+from pipelines.utils.flow import Flow, flow
 
 
 @flow(name="flow_de_teste", log_prints=True)
@@ -47,12 +48,12 @@ def test_atributos_de_deploy_nao_sao_compartilhados():
 
 
 def test_aceita_atribuicao_dos_atributos_de_deploy():
-    schedules: list[DeploySchedule] = [
-        {"cron": "0 16 10 * *", "timezone": "America/Sao_Paulo"}
-    ]
+    schedules = [Cron("0 16 10 * *", timezone="America/Sao_Paulo")]
 
     _flow_com_opcoes.deploy_schedules = schedules
     _flow_com_opcoes.job_variables = {"memory": "8Gi"}
 
     assert _flow_com_opcoes.deploy_schedules == schedules
+    assert _flow_com_opcoes.deploy_schedules[0].cron == "0 16 10 * *"
+    assert _flow_com_opcoes.deploy_schedules[0].timezone == "America/Sao_Paulo"
     assert _flow_com_opcoes.job_variables == {"memory": "8Gi"}
diff --git a/pipelines/{{cookiecutter.pipeline_name}}/flows.py b/pipelines/{{cookiecutter.pipeline_name}}/flows.py
index 986db05071..e04346b4a0 100644
--- a/pipelines/{{cookiecutter.pipeline_name}}/flows.py
+++ b/pipelines/{{cookiecutter.pipeline_name}}/flows.py
@@ -17,7 +17,7 @@
 # storage nem run_config manualmente.
 #
 # O agendamento é definido inline no próprio objeto do flow, via
-# `<flow>.deploy_schedules`, uma lista de dicts `{"cron": ..., "timezone": ...}`.
+# `<flow>.deploy_schedules`, uma lista de `Cron` (`prefect.schedules`).
 # Em `dev` os schedules são ignorados; em `prod` são ativados na sincronização
 # com o backend. Veja https://crontab.guru/ para montar a expressão cron.
 #
@@ -27,6 +27,8 @@
 #
 ###############################################################################
 
+from prefect.schedules import Cron
+
 from pipelines.datasets.{{cookiecutter.pipeline_name}}.tasks import say_hello
 from pipelines.utils.flow import flow
 
@@ -38,5 +40,5 @@ def {{cookiecutter.pipeline_name}}_flow(name: str = "World") -> None:
 
 # Agendamento (opcional). Remova se o flow não deve rodar em intervalos fixos.
 {{cookiecutter.pipeline_name}}_flow.deploy_schedules = [
-    {"cron": "0 14 * * 1", "timezone": "America/Sao_Paulo"}  # segundas, 14:00
+    Cron("0 14 * * 1", timezone="America/Sao_Paulo")  # segundas, 14:00
 ]

From 512008be19494e3e3fc95c2cbd355a1b1704b46f Mon Sep 17 00:00:00 2001
From: Pedro Castro <aspeddro@gmail.com>
Date: Fri, 7 Aug 2026 22:05:50 -0300
Subject: [PATCH 3/3] add None

---
 .github/scripts/deploy_flows.py | 14 ++++++--------
 pipelines/utils/flow.py         |  8 ++++----
 2 files changed, 10 insertions(+), 12 deletions(-)

diff --git a/.github/scripts/deploy_flows.py b/.github/scripts/deploy_flows.py
index 395bb7d65a..6bd2854f67 100644
--- a/.github/scripts/deploy_flows.py
+++ b/.github/scripts/deploy_flows.py
@@ -15,9 +15,10 @@
 import sys
 from pathlib import Path
 
-from prefect import Flow
 from prefect.runner.storage import GitRepository
 
+from pipelines.utils.flow import Flow
+
 REPO_URL = "https://github.com/basedosdados/pipelines.git"
 
 
@@ -68,18 +69,15 @@ def deploy_flow(
     entrypoint = f"{file_path}:{flow_name}"
     is_dev = "dev" in pool_name
 
-    # `deploy_schedules` e `job_variables` são declarados em
-    # `pipelines.utils.flow.Flow` e vêm vazios quando o flow não os define;
-    # `or None` normaliza para o que o Prefect entende como "não informado".
-    schedules = getattr(flow, "deploy_schedules", None) or None
-    if is_dev:
-        schedules = None  # flows em dev não têm schedule
+    # flows em dev não têm schedule
+    schedules = None if is_dev else flow.deploy_schedules
 
-    job_variables = getattr(flow, "job_variables", None) or None
+    job_variables = flow.job_variables
 
     print(f"  Registrando {flow_name} → {entrypoint}")
 
     try:
+        # pyrefly: ignore [missing-attribute]
         flow.from_source(
             source=GitRepository(
                 url=REPO_URL,
diff --git a/pipelines/utils/flow.py b/pipelines/utils/flow.py
index c7f01e2dfe..bc5a6309ae 100644
--- a/pipelines/utils/flow.py
+++ b/pipelines/utils/flow.py
@@ -58,13 +58,13 @@ class Flow(PrefectFlow[P, R]):
             pool, por exemplo `{"memory": "8Gi"}`. Vazio usa o padrão do pool.
     """
 
-    deploy_schedules: list[Schedule]
-    job_variables: dict[str, Any]
+    deploy_schedules: list[Schedule] | None
+    job_variables: dict[str, Any] | None
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
-        self.deploy_schedules = []
-        self.job_variables = {}
+        self.deploy_schedules = None
+        self.job_variables = None
 
 
 # Uso sem parênteses (`@flow`). A assinatura da função decorada não é