chore: Flow customizado com os atributos de deploy

.claude/rules/prefect-pipeline-conventions.md

@@ -32,6 +32,21 @@ 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). `deploy_schedules` is a list of
+`prefect.schedules.Cron` — `Cron("0 16 10 * *", timezone="America/Sao_Paulo")`.
+
 ## DRY with the onboarding code
 
 The cleaning transform lives in **one place** and is shared:
@@ -275,8 +290,17 @@ 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
+
+
+@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"}
+    Cron("0 16 10,11,12,13 * *", timezone="America/Sao_Paulo")
 ]
 my_flow.job_variables = {
     "memory": "8Gi"
@@ -290,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.
 

.github/scripts/deploy_flows.py

@@ -15,9 +15,9 @@
 import sys
 from pathlib import Path
 
-from prefect import Flow
 from prefect.runner.storage import GitRepository
-from prefect.schedules import Cron
+
+from pipelines.utils.flow import Flow
 
 REPO_URL = "https://github.com/basedosdados/pipelines.git"
 
@@ -69,23 +69,15 @@ def deploy_flow(
     entrypoint = f"{file_path}:{flow_name}"
     is_dev = "dev" in pool_name
 
-    schedules = getattr(flow, "deploy_schedules", 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
-        ]
+    # flows em dev não têm schedule
+    schedules = None if is_dev else flow.deploy_schedules
 
-    job_variables = getattr(flow, "job_variables", 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,

AGENTS.md

@@ -64,18 +64,25 @@ 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.
 
-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
-my_flow.deploy_schedules = [
-    {"cron": "0 16 10 * *", "timezone": "America/Sao_Paulo"}
-]
+from prefect.schedules import Cron
+
+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")]
 my_flow.job_variables = {
     "memory": "8Gi"
 }  # optional; size to the flow's peak RAM

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 = []

pipelines/datasets/au_abs_cpi/flows.py

@@ -15,10 +15,11 @@
 import shutil
 import tempfile
 
-from prefect import flow
+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
 from pipelines.utils.metadata.domain import (
     AllFree,
     DateFormat,
@@ -173,7 +174,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"}
+    Cron("0 16 22,23,24,25,26,27,28 * *", timezone="America/Sao_Paulo")
 ]

pipelines/datasets/au_abs_labour_force/flows.py

@@ -19,7 +19,7 @@
 import shutil
 import tempfile
 
-from prefect import flow
+from prefect.schedules import Cron
 
 from pipelines.datasets.au_abs_labour_force.constants import constants
 from pipelines.datasets.au_abs_labour_force.tasks import (
@@ -28,6 +28,7 @@
     download_sdmx_task,
     latest_month_task,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     FreeLag,
@@ -191,10 +192,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"}
+    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"}

pipelines/datasets/br_anatel_banda_larga_fixa/flows.py

@@ -5,11 +5,12 @@
 deste diretório.
 """
 
-from prefect import flow
+from prefect.schedules import Cron
 
 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,9 +40,8 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
     _flow.deploy_schedules = (
-        [{"cron": cron, "timezone": "America/Sao_Paulo"}] if cron else []
+        [Cron(cron, timezone="America/Sao_Paulo")] if cron else []
     )
     return _flow
 

pipelines/datasets/br_anatel_telefonia_movel/flows.py

@@ -1,10 +1,11 @@
 """Flows for br_anatel_telefonia_movel — Prefect 3."""
 
-from prefect import flow
+from prefect.schedules import Cron
 
 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 +36,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.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     _flow.job_variables = {"memory_limit": "8Gi", "memory_request": "2Gi"}
     return _flow
 

pipelines/datasets/br_anp_precos_combustiveis/flows.py

@@ -2,13 +2,14 @@
 Flow br_anp_precos_combustiveis__microdados — Prefect 3.
 """
 
-from prefect import flow
+from prefect.schedules import Cron
 
 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 +121,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"}
+    Cron("0 10 * * *", timezone="America/Sao_Paulo")
 ]

pipelines/datasets/br_ans_beneficiario/flows.py

@@ -2,14 +2,15 @@
 Flow br_ans_beneficiario__informacao_consolidada — Prefect 3.
 """
 
-from prefect import flow
+from prefect.schedules import Cron
 
 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 +134,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"}
+    Cron("0 21 * * *", timezone="America/Sao_Paulo")
 ]

pipelines/datasets/br_bcb_agencia/flows.py

@@ -2,7 +2,7 @@
 Flow br_bcb_agencia__agencia — Prefect 3.
 """
 
-from prefect import flow
+from prefect.schedules import Cron
 
 from pipelines.crawler.bcb_agencia.tasks import (
     clean_data,
@@ -11,6 +11,7 @@
     get_documents_metadata,
     get_latest_file,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     PartBdpro,
@@ -146,7 +147,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"}
+    Cron("0 22 25-31 * *", timezone="America/Sao_Paulo")
 ]

pipelines/datasets/br_bcb_estban/flows.py

@@ -2,7 +2,7 @@
 Flows para br_bcb_estban — Prefect 3.
 """
 
-from prefect import flow
+from prefect.schedules import Cron
 
 from pipelines.crawler.bcb_estban.tasks import (
     cleaning_data,
@@ -12,6 +12,7 @@
     get_id_municipio,
     get_latest_file,
 )
+from pipelines.utils.flow import flow
 from pipelines.utils.metadata.domain import (
     DateFormat,
     PartBdpro,
@@ -167,8 +168,7 @@ def _flow(
             force_run=force_run,
         )
 
-    # pyrefly: ignore [missing-attribute]
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 

pipelines/datasets/br_bcb_sicor/flows.py

@@ -2,10 +2,11 @@
 Flows para br_bcb_sicor — Prefect 3.
 """
 
-from prefect import flow
+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
 from pipelines.utils.tasks import (
     rename_flow_run_dataset_table,
     run_dbt,
@@ -52,8 +53,7 @@ def _flow(
             local_redis_execution=local_redis_execution,
         )
 
-    # pyrefly: ignore [missing-attribute]
-    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    _flow.deploy_schedules = [Cron(cron, timezone="America/Sao_Paulo")]
     return _flow
 
 

pipelines/datasets/br_bcb_taxa_cambio/flows.py

@@ -2,12 +2,13 @@
 Flow br_bcb_taxa_cambio — Prefect 3.
 """
 
-from prefect import flow
+from prefect.schedules import Cron
 
 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 +95,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"}
+    Cron("0 8 * * *", timezone="America/Sao_Paulo")
 ]