From f2654e5367a1497db8ffca4c817d8d1c1ab2626b Mon Sep 17 00:00:00 2001 From: Ricardo Dahis Date: Tue, 1 Sep 2026 16:35:06 +1000 Subject: [PATCH] fix(utils): _sync_staging_schema usa o cliente autenticado da staging bigquery.Client(project=...) sem credencial cai no ADC do pod, que nao e o principal com acesso a staging: get_table estoura 403 em dataset novo e update_table estoura 403 sempre. O cliente correto ja vem no proprio objeto bd.Table recebido pela funcao. Derrubou o backfill de br_senatran_estatisticas.municipio_combustivel duas vezes (109 meses, ~47 min cada) e estava documentado como pendencia no README do br_sfb_sicar. --- pipelines/datasets/br_sfb_sicar/README.md | 10 ++++------ pipelines/utils/tasks.py | 9 +++++---- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/pipelines/datasets/br_sfb_sicar/README.md b/pipelines/datasets/br_sfb_sicar/README.md index d2a7f36c23..54d066375e 100644 --- a/pipelines/datasets/br_sfb_sicar/README.md +++ b/pipelines/datasets/br_sfb_sicar/README.md @@ -118,12 +118,10 @@ sem nenhum modelo lendo essa coluna — e derruba depois do download inteiro. ## Pendências -- **O PATCH do schema da staging falha com 403.** `_sync_staging_schema` - (`pipelines/utils/tasks.py`) abre `bigquery.Client(project=...)` sem credencial e cai no - ADC do pod, que só lê; o `get_table` passa e o `update_table` estoura com - `bigquery.tables.update denied`. O cliente autenticado está em - `tb.client["bigquery_staging"]`, no objeto que a função já recebe. Vale para este e para - qualquer outro conjunto; o conserto sai em PR à parte. +- ~~**O PATCH do schema da staging falha com 403.**~~ Corrigido: + `_sync_staging_schema` (`pipelines/utils/tasks.py`) usava + `bigquery.Client(project=...)` sem credencial e caía no ADC do pod, que só lê. Agora usa + `tb.client["bigquery_staging"]`, o cliente autenticado que já vem no objeto recebido. - **Não há staging em dev.** Nem o dataset `br_sfb_sicar_staging` em `basedosdados-dev`, nem o prefixo `gs://basedosdados-dev/staging/br_sfb_sicar/`. O próximo run no pool de dev os cria pelo ramo `tb.create` do `upload_to_gcs`. diff --git a/pipelines/utils/tasks.py b/pipelines/utils/tasks.py index 5772f1b8c3..e1a302b881 100644 --- a/pipelines/utils/tasks.py +++ b/pipelines/utils/tasks.py @@ -76,7 +76,6 @@ def _sync_staging_schema( tb: bd.Table, data_path: str | Path, source_format: str, - billing_project_id: str, ) -> None: """Adiciona ao schema da staging as colunas que a fonte passou a trazer. @@ -103,14 +102,17 @@ def _sync_staging_schema( tb: tabela `basedosdados` já instanciada, apontando para a staging. data_path: arquivo ou diretório com os dados que serão carregados. source_format: `"csv"` ou `"parquet"`. - billing_project_id: projeto GCP usado para faturar a chamada. """ header_path = dump_header(data_path=data_path, source_format=source_format) incoming = tb._load_staging_schema_from_data( data_sample_path=header_path, source_format=source_format ) - client = bigquery.Client(project=billing_project_id) + # O cliente tem que ser o da própria lib: `bigquery.Client()` sem + # credencial cai no ADC do pod, que não é o principal com acesso à + # staging — o `get_table` estoura 403 em dataset novo e o + # `update_table` estoura 403 sempre. + client = tb.client["bigquery_staging"] table = client.get_table(tb.table_full_name["staging"]) current = {_bq_safe_column_name(field.name) for field in table.schema} @@ -190,7 +192,6 @@ def _upload_to_gcs( tb=tb, data_path=data_path, source_format=source_format, - billing_project_id=billing_project_id, ) elif dump_mode == "overwrite":