chore: Flow customizado com os atributos de deploy - #1771
Draft
aspeddro wants to merge 3 commits into
Draft
Conversation
`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>
📝 WalkthroughWalkthroughThis change adds a Prefect-compatible project ChangesCustom Flow Wrapper Migration
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant PipelineFlow
participant ProjectFlow
participant PrefectFlow
participant DeployScript
PipelineFlow->>ProjectFlow: apply `@flow`
ProjectFlow->>PrefectFlow: construct compatible Flow
PrefectFlow-->>PipelineFlow: expose deploy_schedules and job_variables
DeployScript->>PipelineFlow: read deployment attributes
DeployScript->>PrefectFlow: register normalized deployment values
Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 3📝 Generate docstrings 💡
⚔️ Resolve merge conflicts 💡
🛠️ Fix failing CI checks 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
|
@aspeddro esse pull request tem conflitos 😩 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
pipelines/utils/flow.py (1)
87-91: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winPreserve the decorated signature for
@flow.The direct decorator overload currently maps
Callable[..., Any]toFlow[..., Any], so direct@flowuse loses the flow’s parameter and return types. Use the existingPandRtype variables for this overload and fordecorator(...). Prefect 3 already modelsFlowandflowwithParamSpec["P"]andReturnType["R"].🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/utils/flow.py` around lines 87 - 91, Update the direct `@flow` overload to use the existing P and R type variables, preserving the decorated callable’s parameter and return types in its Flow result. Apply the same P/R typing consistently to decorator(...), while retaining the current positional-only signature and avoiding a new type-parameter syntax.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@AGENTS.md`:
- Line 67: Clarify the flow-authoring rule in AGENTS.md: exported Flow objects
must be discoverable from flows.py and their underlying function must be defined
in that file, while factory functions such as br_ibge_ipca may still define
valid nested flow functions there. Align the wording with
prefect-pipeline-conventions.md without changing the decorator requirement.
In `@pipelines/utils/flow.py`:
- Around line 81-84: Add Google-style docstrings to Flow.__init__ and decorator
in pipelines/utils/flow.py, documenting their parameters and behavior; add
Google-style docstrings to the helper and test functions in
pipelines/utils/tests/test_flow.py, and annotate every test function with ->
None. Update all three listed sites: pipelines/utils/flow.py lines 81-84 and
134-135, and pipelines/utils/tests/test_flow.py lines 8-58.
In `@pipelines/utils/tests/test_flow.py`:
- Around line 49-58: Update test_aceita_atribuicao_dos_atributos_de_deploy so it
does not mutate the shared module-level _flow_com_opcoes; create an isolated
flow instance with the same configuration inside the test, or restore
deploy_schedules and job_variables in a finally block while preserving the
existing assertions.
---
Nitpick comments:
In `@pipelines/utils/flow.py`:
- Around line 87-91: Update the direct `@flow` overload to use the existing P and
R type variables, preserving the decorated callable’s parameter and return types
in its Flow result. Apply the same P/R typing consistently to decorator(...),
while retaining the current positional-only signature and avoiding a new
type-parameter syntax.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 19f55518-0f33-4935-9227-2f90666f6c2a
📒 Files selected for processing (65)
.claude/rules/prefect-pipeline-conventions.md.github/scripts/deploy_flows.pyAGENTS.mdpipelines/crawler/ibge_inflacao/flows.pypipelines/datasets/au_abs_cpi/flows.pypipelines/datasets/au_abs_labour_force/flows.pypipelines/datasets/br_anatel_banda_larga_fixa/flows.pypipelines/datasets/br_anatel_telefonia_movel/flows.pypipelines/datasets/br_anp_precos_combustiveis/flows.pypipelines/datasets/br_ans_beneficiario/flows.pypipelines/datasets/br_bcb_agencia/flows.pypipelines/datasets/br_bcb_estban/flows.pypipelines/datasets/br_bcb_sicor/flows.pypipelines/datasets/br_bcb_taxa_cambio/flows.pypipelines/datasets/br_bcb_taxa_selic/flows.pypipelines/datasets/br_bd_indicadores/flows.pypipelines/datasets/br_bd_siga_o_dinheiro/flows.pypipelines/datasets/br_bndes_operacoes_contratadas/flows.pypipelines/datasets/br_camara_dados_abertos/flows.pypipelines/datasets/br_cgu_beneficios_cidadao/flows.pypipelines/datasets/br_cgu_cartao_pagamento/flows.pypipelines/datasets/br_cgu_emendas_parlamentares/flows.pypipelines/datasets/br_cgu_licitacao_contrato/flows.pypipelines/datasets/br_cgu_pessoal_executivo_federal/flows.pypipelines/datasets/br_cgu_servidores_executivo_federal/flows.pypipelines/datasets/br_cnj_improbidade_administrativa/flows.pypipelines/datasets/br_cvm_administradores_carteira/flows.pypipelines/datasets/br_cvm_fi/flows.pypipelines/datasets/br_cvm_oferta_publica_distribuicao/flows.pypipelines/datasets/br_denatran_frota/flows.pypipelines/datasets/br_fgv_igp/flows.pypipelines/datasets/br_ibge_inpc/flows.pypipelines/datasets/br_ibge_ipca/flows.pypipelines/datasets/br_ibge_ipca15/flows.pypipelines/datasets/br_ibge_pnadc/flows.pypipelines/datasets/br_inmet_bdmep/flows.pypipelines/datasets/br_me_caged/flows.pypipelines/datasets/br_me_cnpj/flows.pypipelines/datasets/br_me_comex_stat/flows.pypipelines/datasets/br_me_rais/flows.pypipelines/datasets/br_me_siconfi/flows.pypipelines/datasets/br_mp_pep/flows.pypipelines/datasets/br_ms_cnes/flows.pypipelines/datasets/br_ms_sia/flows.pypipelines/datasets/br_ms_sih/flows.pypipelines/datasets/br_ms_sinan/flows.pypipelines/datasets/br_poder360_pesquisas/flows.pypipelines/datasets/br_rf_cafir/flows.pypipelines/datasets/br_rf_cno/flows.pypipelines/datasets/br_rj_isp_estatisticas_seguranca/flows.pypipelines/datasets/br_senado_dados_abertos/flows.pypipelines/datasets/br_sfb_sicar/flows.pypipelines/datasets/br_stf_corte_aberta/flows.pypipelines/datasets/br_tse_eleicoes/flows.pypipelines/datasets/fundacao_lemann/flows.pypipelines/datasets/test_dataset/flows.pypipelines/datasets/us_bls_cpi/flows.pypipelines/datasets/us_bls_qcew/flows.pypipelines/datasets/world_cricsheet/flows.pypipelines/utils/execute_dbt_model/flows.pypipelines/utils/flow.pypipelines/utils/materialize_prod/flows.pypipelines/utils/metadata/flows.pypipelines/utils/tests/test_flow.pypipelines/{{cookiecutter.pipeline_name}}/flows.py
| ### 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). |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Clarify the module-level flow rule.
Line 67 says that flows must be defined at module level. .claude/rules/prefect-pipeline-conventions.md permits a factory such as br_ibge_ipca when the inner function is defined in the same file. State that the exported Flow object must be discoverable from flows.py, while nested factory functions remain valid.
The repository guides should not give conflicting authoring rules.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@AGENTS.md` at line 67, Clarify the flow-authoring rule in AGENTS.md: exported
Flow objects must be discoverable from flows.py and their underlying function
must be defined in that file, while factory functions such as br_ibge_ipca may
still define valid nested flow functions there. Align the wording with
prefect-pipeline-conventions.md without changing the decorator requirement.
Comment on lines
+81
to
+84
| def __init__(self, *args: Any, **kwargs: Any) -> None: | ||
| super().__init__(*args, **kwargs) | ||
| self.deploy_schedules = [] | ||
| self.job_variables = {} |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Add the required function documentation and annotations.
pipelines/utils/flow.py#L81-L84: Add a Google-style docstring forFlow.__init__.pipelines/utils/flow.py#L134-L135: Add a Google-style docstring fordecorator.pipelines/utils/tests/test_flow.py#L8-L58: Add Google-style docstrings to the helper and test functions. Add-> Noneto each test function.
As per coding guidelines, “add Google-Style type hints and docstrings to Python functions.”
📍 Affects 2 files
pipelines/utils/flow.py#L81-L84(this comment)pipelines/utils/flow.py#L134-L135pipelines/utils/tests/test_flow.py#L8-L58
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pipelines/utils/flow.py` around lines 81 - 84, Add Google-style docstrings to
Flow.__init__ and decorator in pipelines/utils/flow.py, documenting their
parameters and behavior; add Google-style docstrings to the helper and test
functions in pipelines/utils/tests/test_flow.py, and annotate every test
function with -> None. Update all three listed sites: pipelines/utils/flow.py
lines 81-84 and 134-135, and pipelines/utils/tests/test_flow.py lines 8-58.
Source: Coding guidelines
Comment on lines
+49
to
+58
| 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"} |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Avoid mutation of a module-level test flow.
This test changes _flow_com_opcoes without restoring its defaults. Test order changes can then make tests that expect empty attributes fail. Create a flow inside this test, or restore both attributes in finally.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pipelines/utils/tests/test_flow.py` around lines 49 - 58, Update
test_aceita_atribuicao_dos_atributos_de_deploy so it does not mutate the shared
module-level _flow_com_opcoes; create an isolated flow instance with the same
configuration inside the test, or restore deploy_schedules and job_variables in
a finally block while preserving the existing assertions.
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>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
deploy_flows.pylêdeploy_schedulesejob_variablesdo objeto flow, masprefect.Flownão declara nenhum dos dois — o Pyrefly acusamissing-attribute, hoje silenciado com 80# pyrefly: ignorenosflows.py.Adiciona
pipelines/utils/flow.py: subclasse deprefect.Flowque declara os dois atributos (vazios por padrão) e um decoratorflowque a instancia, com as mesmas opções doprefect.flow. Migra os 60flows.pye remove as supressões.Verificação:
pyrefly checksem diagnósticos (972 → 892 supressões) e o loader do CI descobre os mesmos 189 flows, com schedules ejob_variablesidênticos, antes e depois.Exemplo de uso
🤖 Generated with Claude Code