[Data] br_rf_cnpj: correção do tabble-approve - #1798
Merged
Conversation
…râmetro de download_max_parallel de 5 para 15 (era o que estava sendo manualmente inserido nas runs), mudança no padrão de datas no registro de metadados, mudança na query de br_bd_diretorios_brasil__empresa para usar br_rf_cnpj ao invés de br_me_cnpj. Também correções em arquivos com erros do Pyrefly
luizavboas
added
table-approve
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughThe PR migrates the directory company model to RF CNPJ snapshots, updates CNPJ crawler date handling and download concurrency, adds five scheduled Prefect flows, suppresses Pyrefly diagnostics, and completes a JSON diagnostics file. ChangesCNPJ model and pipeline
Static analysis and diagnostics
Estimated code review effort: 3 (Moderate) | ~20 minutes Mergeability Score: ⚪ Minimal · up to The PR updates dataset references, metadata date formats, approval comments, and a download parallelism default; no actionable merge-blocking risk remains based on the supplied evidence. Sequence Diagram(s)sequenceDiagram
participant PrefectDeployment
participant DatasetFlowFactory
participant _run_rf_cnpj
PrefectDeployment->>DatasetFlowFactory: trigger scheduled RF CNPJ flow
DatasetFlowFactory->>_run_rf_cnpj: forward table and loading options
_run_rf_cnpj-->>PrefectDeployment: complete table run
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 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 `@models/br_bd_diretorios_brasil/br_bd_diretorios_brasil__empresa.sql`:
- Around line 161-162: Update the join condition in the empresa query to match
the `situacao_cadastral` dictionary key against the establishment’s
`situacao_cadastral` field, not `identificador_matriz_filial`. Keep the existing
dictionary alias `t` and other query logic unchanged.
In `@models/br_rf_cnpj/br_rf_cnpj__socios_legado.sql`:
- Around line 1-2: Remove the empty br_rf_cnpj__socios_legado model if it is
unused; otherwise, implement executable SQL in the model and add its
corresponding schema.yml metadata so it is a valid active dbt model.
In `@pipelines/crawler/rf_cnpj/tasks.py`:
- Line 57: Document the download_max_parallel parameter in main’s Google-style
Args section, including its purpose and default value of 15. Follow the existing
Python docstring and type-hint conventions without changing the parameter
behavior.
- Line 57: Update the non-segmented download call in the task flow to pass the
configured download_max_parallel value into download_unzip_csv, matching the
segmented-table branch and preserving the flow configuration for all downloads.
In `@pipelines/datasets/flows.py`:
- Around line 10-29: Document both `_rf_cnpj_flow` and its nested `_flow` with
Google-style docstrings describing their parameters, schedule behavior, and
return values; add the appropriate return type annotation to `_rf_cnpj_flow`.
Preserve the existing flow configuration and parameter defaults while
documenting the public factory contract.
🪄 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: 19c97714-2c10-4719-981c-be749ed44164
📒 Files selected for processing (19)
models/br_bd_diretorios_brasil/br_bd_diretorios_brasil__empresa.sqlmodels/br_rf_cnpj/br_rf_cnpj__dicionario.sqlmodels/br_rf_cnpj/br_rf_cnpj__empresas.sqlmodels/br_rf_cnpj/br_rf_cnpj__empresas_legado.sqlmodels/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sqlmodels/br_rf_cnpj/br_rf_cnpj__estabelecimentos_legado.sqlmodels/br_rf_cnpj/br_rf_cnpj__simples.sqlmodels/br_rf_cnpj/br_rf_cnpj__socios._legado.sqlmodels/br_rf_cnpj/br_rf_cnpj__socios.sqlmodels/br_rf_cnpj/br_rf_cnpj__socios_legado.sqlmodels/us_harvard_cbdb/code/clean.pymodels/us_harvard_cbdb/code/gen_architecture.pymodels/us_harvard_cbdb/code/gen_dbt.pymodels/us_harvard_cbdb/code/gen_metadata_payload.pymodels/us_harvard_cbdb/code/schema_spec.pymodels/us_harvard_cbdb/code/upload.pypipelines/crawler/rf_cnpj/flows.pypipelines/crawler/rf_cnpj/tasks.pypipelines/datasets/flows.py
Comment on lines
+1
to
+2
| -- Atualizado em 2026-08-11 | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- Model contents ---'
sed -n '1,80p' models/br_rf_cnpj/br_rf_cnpj__socios_legado.sql
printf '%s\n' '--- dbt project configuration ---'
fd -HI 'dbt_project\.yml' . -x sed -n '1,240p' {}
printf '%s\n' '--- References and selectors ---'
rg -n -C 3 'br_rf_cnpj__socios_legado|socios_legado|br_rf_cnpj' \
-g '*.sql' -g '*.yml' -g '*.yaml' -g '*.md' .Repository: basedosdados/pipelines
Length of output: 24118
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
import re
model = Path("models/br_rf_cnpj/br_rf_cnpj__socios_legado.sql")
project = Path("dbt_project.yml")
schema = Path("models/br_rf_cnpj/schema.yml")
content = model.read_text()
executable = "\n".join(
line for line in content.splitlines()
if line.strip() and not line.lstrip().startswith("--")
)
print(f"model_exists={model.exists()}")
print(f"model_path={model}")
print(f"executable_sql={bool(executable.strip())}")
print(f"configured_model_path={'model-paths: [models]' in project.read_text()}")
print(
"dataset_configured="
+ str(bool(re.search(r"(?m)^ br_rf_cnpj:\s*$", project.read_text())))
)
print(
"schema_entry="
+ str(bool(re.search(
r"(?m)^\s*-\s+name:\s+br_rf_cnpj__socios_legado\s*$",
schema.read_text(),
)))
)
PYRepository: basedosdados/pipelines
Length of output: 328
Remove or implement this dbt model.
models/br_rf_cnpj/br_rf_cnpj__socios_legado.sql contains no executable SQL, is included in the active models path, and has no schema.yml entry. Delete it if unused. Otherwise, add the model query and metadata.
🤖 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 `@models/br_rf_cnpj/br_rf_cnpj__socios_legado.sql` around lines 1 - 2, Remove
the empty br_rf_cnpj__socios_legado model if it is unused; otherwise, implement
executable SQL in the model and add its corresponding schema.yml metadata so it
is a valid active dbt model.
| download_chunk_size: int = 15 * 1024 * 1024, | ||
| download_max_retries: int = 5, | ||
| download_max_parallel: int = 5, | ||
| download_max_parallel: int = 15, |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Document download_max_parallel in main.
Add this parameter to the Google-style Args section. Its default changed to 15, but callers cannot find its purpose or default in the function documentation.
As per coding guidelines: **/*.py must use Google-Style type hints and docstrings.
🤖 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/crawler/rf_cnpj/tasks.py` at line 57, Document the
download_max_parallel parameter in main’s Google-style Args section, including
its purpose and default value of 15. Follow the existing Python docstring and
type-hint conventions without changing the parameter behavior.
Source: Coding guidelines
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
Pass the configured download controls to non-segmented downloads.
The new default reaches download_unzip_csv only in the segmented-table branch. The call at Line 161 omits max_parallel, so non-segmented tables continue to use download_unzip_csv defaults instead of the flow configuration.
Proposed fix
- asyncio.run(download_unzip_csv(url_download, input_path))
+ asyncio.run(
+ download_unzip_csv(
+ url_download,
+ input_path,
+ chunk_size=download_chunk_size,
+ max_retries=download_max_retries,
+ max_parallel=download_max_parallel,
+ timeout=download_timeout,
+ )
+ )🤖 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/crawler/rf_cnpj/tasks.py` at line 57, Update the non-segmented
download call in the task flow to pass the configured download_max_parallel
value into download_unzip_csv, matching the segmented-table branch and
preserving the flow configuration for all downloads.
laura-l-amaral
approved these changes
Aug 12, 2026
Contributor
|
Tick the box to add this pull request to the merge queue (same as
|
Contributor
|
@luizavboas esse pull request tem conflitos 😩 |
Winzen
added a commit
that referenced
this pull request
Aug 13, 2026
O #1783 moveu commit_source_update_task pra logo após o poll, removendo a chamada antiga do fim do flow em todos os arquivos afetados. O #1798, mergeado depois mas cortado de uma base anterior a esse merge, ainda tinha essa chamada antiga e só editou parâmetros dela (date_format). O merge do git combinou as duas sem detectar a duplicação lógica — commit_source_update_task passou a ser chamado duas vezes por run. Sem impacto de dados (é idempotente), só uma escrita redundante.
Winzen
added a commit
that referenced
this pull request
Aug 14, 2026
…a em rf_cnpj (#1779) A migração pro poll.py que este PR propunha originalmente não é mais necessária: o #1783 já corrigiu poll_source_for_update via compare_against, e br_ans_beneficiario já usa esse default corrigido. Mantido só o que continua valendo independente do mecanismo de poll: - br_ans_beneficiario: colunas de texto categóricas trocadas de str para category (reduz footprint de memória do DataFrame), del/gc.collect() por estado em parquet_partition (corrige OOM), source_format="parquet" nos upload_to_gcs (sem isso o dump_header procurava .csv e não achava nada), job_variables={"memory": "3Gi"} (pico medido ~1.78Gi depois da otimização), compare_against="coverage" explícito no poll. - rf_cnpj: remove uma chamada duplicada de commit_source_update_task que sobrou de um merge entre o #1783 (moveu a chamada pra cima) e o #1798 (mergeado depois, cortado de uma base anterior, ainda tinha a chamada antiga) — sem impacto de dados (idempotente), só uma escrita redundante. - Exclui models/world_aiddata_gcdf/code do Pyrefly (mesmo padrão de br_tse_eleicoes/us_harvard_cbdb/us_cfpb_hmda) — destrava o CI deste PR.
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
2 participants
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.
Descrição do PR:
Correções para o table-approve do
br_rf_cnpje ajustes de pipeline associados.br_rf_cnpjestava referenciado incorretamente embr_bd_diretorios_brasil__empresa.sql(ainda apontava para obr_me_cnpjantigo), os modelos SQL não tinham os comentários exigidos pelo table-approve, e o parâmetrodownload_max_paralleldivergia do valor real usado nas runs manuais.Detalhes Técnicos:
br_bd_diretorios_brasil__empresa.sql: migrado debr_me_cnpjparabr_rf_cnpj, trocando a coluna de data dedataparadata_referencia.Adicionado comentário
-- Atualizado em 2026-08-11em todos os models debr_rf_cnpjpara o table-approve;pipelines/crawler/rf_cnpj/flows.py: deDateFormat.YEAR_MDparaDateFormat.YEAR_MONTHno registro de metadados (empresas/estabelecimentos/socios e diretórioempresa), edate_format="%Y-%m"padronizado para o enumDateFormat.YEAR_MONTH.pipelines/crawler/rf_cnpj/tasks.py:download_max_paralleldefault alterado de5para15, alinhando ao valor já usado manualmente nas execuções.Comentários
# pyrefly: ignore [...]para suprimir diagnósticos indesejados do PyreflySummary by CodeRabbit