[Bugfix] _upload_to_gcs nunca pode apagar a tabela de produção

pipelines/utils/tasks.py

@@ -72,11 +72,34 @@ def _bq_safe_column_name(name: str) -> str:
     return re.sub(r"[^0-9a-zA-Z_]", "_", name)
 
 
+def _staging_client(tb: bd.Table) -> bigquery.Client:
+    """Devolve o cliente BigQuery que a própria `bd.Table` usa para a staging.
+
+    Nunca construa um `bigquery.Client()` novo para falar com a staging. O
+    cliente novo cai no ADC do pod, que não tem `bigquery.tables.get` nas
+    tabelas de staging; a lib carrega `BASEDOSDADOS_CREDENTIALS_STAGING`, que
+    tem. Os dois convivem no mesmo processo, então o sintoma é enganoso: em
+    2026-08-21 o flow `us_treasury_usaspending` (run `nano-cheetah`) logou
+    "Tabela já existe" — `tb.table_exists()`, cliente da lib — e um segundo
+    depois levou 403 `Permission bigquery.tables.get denied` na mesma tabela,
+    vindo de um `bigquery.Client()` construído à mão.
+
+    Só morde `dump_mode="append"`: o ramo `overwrite` não chama nenhuma das
+    duas funções de sync, e por isso o defeito passou despercebido.
+
+    Args:
+        tb: tabela `basedosdados` já instanciada, apontando para a staging.
+
+    Returns:
+        O cliente BigQuery autenticado com as credenciais de staging.
+    """
+    return tb.client["bigquery_staging"]
+
+
 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 +126,13 @@ 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)
+    client = _staging_client(tb)
     table = client.get_table(tb.table_full_name["staging"])
 
     current = {_bq_safe_column_name(field.name) for field in table.schema}
@@ -138,6 +160,68 @@ def _sync_staging_schema(
     )
 
 
+def _sync_staging_source_uris(
+    tb: bd.Table,
+    bucket_name: str,
+    dataset_id: str,
+    table_id: str,
+) -> None:
+    """Reaponta a tabela externa de staging para o bucket do ambiente corrente.
+
+    A definição da tabela externa guarda o bucket que estava em uso quando ela
+    foi criada. Como `dump_mode="append"` só cria a tabela quando ela ainda não
+    existe, um bucket errado gravado uma vez fica gravado para sempre: as
+    execuções seguintes escrevem os blobs no bucket certo e o dbt continua lendo
+    o errado, silenciosamente.
+
+    Isso não é hipotético. Em 2026-08-19 a metade dev de um flow recriou
+    `basedosdados-staging.us_sec_edgar_staging.dicionario` apontando para
+    `gs://basedosdados-dev/...`, divergindo das quatro tabelas irmãs, e a
+    materialização de produção passou a ser construída a partir de blobs do
+    bucket de dev. Nenhum log de erro, nenhuma task falha.
+
+    A correção é feita no lugar, pela API do BigQuery, e é idempotente: quando a
+    URI já está correta a função não faz nada.
+
+    Args:
+        tb: tabela `basedosdados` já instanciada, apontando para a staging.
+        bucket_name: bucket do ambiente corrente.
+        dataset_id: `gcp_dataset_id` sem o sufixo `_staging`.
+        table_id: slug da tabela.
+    """
+    suffix = f"/staging/{dataset_id}/{table_id}/*"
+    expected = f"gs://{bucket_name}{suffix}"
+
+    client = _staging_client(tb)
+    table = client.get_table(tb.table_full_name["staging"])
+
+    config = table.external_data_configuration
+    if config is None:
+        return
+
+    current = list(config.source_uris or [])
+    if current == [expected]:
+        return
+
+    # Conservador de propósito: só corrige o caso em que a ÚNICA diferença é o
+    # bucket. Uma tabela com várias URIs, ou com um caminho fora da convenção,
+    # foi montada à mão por alguém — reescrevê-la seria trocar um estrago
+    # silencioso por outro. Nesses casos apenas avisa.
+    if len(current) != 1 or not current[0].endswith(suffix):
+        print(
+            f"AVISO: staging {tb.table_full_name['staging']} tem URIs fora da "
+            f"convenção ({current}); esperado {[expected]}. Não alterado — "
+            "verifique manualmente."
+        )
+        return
+
+    config.source_uris = [expected]
+    table.external_data_configuration = config
+    client.update_table(table, ["external_data_configuration"])
+
+    print(f"URI da staging corrigida: {current} -> {[expected]}")
+
+
 def _upload_to_gcs(
     data_path: str | Path,
     dataset_id: str,
@@ -186,19 +270,30 @@ def _upload_to_gcs(
             )
         else:
             print(f"Tabela já existe: {tb.table_full_name['staging']}")
+            _sync_staging_source_uris(
+                tb=tb,
+                bucket_name=bucket_name,
+                dataset_id=dataset_id,
+                table_id=table_id,
+            )
             _sync_staging_schema(
                 tb=tb,
                 data_path=data_path,
                 source_format=source_format,
-                billing_project_id=billing_project_id,
             )
 
     elif dump_mode == "overwrite":
         if tb.table_exists(mode="staging"):
             st.delete_table(
                 mode="staging", bucket_name=bucket_name, not_found_ok=True
             )
-            tb.delete(mode="all")
+            # mode="staging", NUNCA mode="all": `all` percorre staging E prod,
+            # apagando a tabela MATERIALIZADA de produção. E isso dispara na
+            # iteração dev do laço de ambientes também, porque `bd.Table`
+            # resolve os projetos do BigQuery pelo config do worker, não pelo
+            # `bucket_name` — então uma execução com materialize_to_prod=False
+            # apagava a tabela de produção e retornava antes de reconstruí-la.
+            tb.delete(mode="staging")
             print(f"Tabela anterior removida: {tb.table_full_name['staging']}")
         header_path = dump_header(
             data_path=data_path, source_format=source_format

pipelines/utils/tests/test_upload_to_gcs_safety.py

@@ -0,0 +1,117 @@
+"""Garantias de que `_upload_to_gcs` nunca toca a tabela de produção.
+
+Três estragos silenciosos já aconteceram por causa do que estes testes fixam:
+
+1. `dump_mode="overwrite"` chamava `tb.delete(mode="all")`, e `all` percorre
+   staging E prod — apagando a tabela materializada de produção, inclusive a
+   partir da iteração dev do laço de ambientes.
+2. A tabela externa de staging guardava o bucket usado na criação. Como o ramo
+   `append` só cria a tabela quando ela não existe, um bucket errado gravado uma
+   vez ficava gravado para sempre, e o dbt de produção passava a ler blobs de
+   dev sem nenhum erro.
+3. As duas funções de sync falavam com a staging por um `bigquery.Client()`
+   construído à mão, que cai no ADC do pod e leva 403 — enquanto a `bd.Table`
+   ao lado, com as credenciais de staging da lib, lia a mesma tabela sem
+   problema.
+"""
+
+from unittest.mock import MagicMock, patch
+
+from pipelines.utils.tasks import (
+    _staging_client,
+    _sync_staging_source_uris,
+    _upload_to_gcs,
+)
+
+STAGING = "basedosdados-staging.us_sec_edgar_staging.dicionario"
+
+
+def _bd_table(uris=None):
+    """`bd.Table` de mentira cujo cliente de staging devolve `uris`."""
+    tb = MagicMock()
+    tb.table_full_name = {"staging": STAGING}
+    table = MagicMock()
+    table.external_data_configuration.source_uris = uris
+    tb.client = {"bigquery_staging": MagicMock()}
+    tb.client["bigquery_staging"].get_table.return_value = table
+    return tb
+
+
+def test_staging_client_is_the_libs_not_a_fresh_one():
+    """O ponto do 403: o cliente tem de vir da `bd.Table`, não do ADC."""
+    tb = _bd_table()
+    assert _staging_client(tb) is tb.client["bigquery_staging"]
+
+
+def test_repoints_when_only_the_bucket_differs():
+    tb = _bd_table(["gs://basedosdados-dev/staging/us_sec_edgar/dicionario/*"])
+
+    _sync_staging_source_uris(
+        tb=tb,
+        bucket_name="basedosdados",
+        dataset_id="us_sec_edgar",
+        table_id="dicionario",
+    )
+
+    client = tb.client["bigquery_staging"]
+    client.update_table.assert_called_once()
+    updated, fields = client.update_table.call_args[0]
+    assert fields == ["external_data_configuration"]
+    assert updated.external_data_configuration.source_uris == [
+        "gs://basedosdados/staging/us_sec_edgar/dicionario/*"
+    ]
+
+
+def test_noop_when_already_correct():
+    tb = _bd_table(["gs://basedosdados/staging/us_sec_edgar/dicionario/*"])
+
+    _sync_staging_source_uris(
+        tb=tb,
+        bucket_name="basedosdados",
+        dataset_id="us_sec_edgar",
+        table_id="dicionario",
+    )
+
+    tb.client["bigquery_staging"].update_table.assert_not_called()
+
+
+def test_leaves_non_conventional_uris_alone():
+    """Várias URIs, ou caminho fora da convenção: avisa e não mexe."""
+    tb = _bd_table(
+        [
+            "gs://outro/caminho/custom/*",
+            "gs://outro/caminho/extra/*",
+        ]
+    )
+
+    _sync_staging_source_uris(
+        tb=tb,
+        bucket_name="basedosdados",
+        dataset_id="us_sec_edgar",
+        table_id="dicionario",
+    )
+
+    tb.client["bigquery_staging"].update_table.assert_not_called()
+
+
+@patch("pipelines.utils.tasks.dump_header")
+@patch("pipelines.utils.tasks.bd")
+def test_overwrite_never_deletes_prod(bd_mod, dump_header_mock):
+    """O ponto central: `overwrite` só pode apagar staging."""
+    dump_header_mock.return_value = "/tmp/header.parquet"
+    tb = bd_mod.Table.return_value
+    tb.table_full_name = {"staging": STAGING}
+    tb.table_exists.return_value = True
+
+    _upload_to_gcs(
+        data_path="/tmp/data",
+        dataset_id="us_sec_edgar",
+        table_id="dicionario",
+        bucket_name="basedosdados-dev",
+        dump_mode="overwrite",
+        source_format="parquet",
+    )
+
+    tb.delete.assert_called_once_with(mode="staging")
+    for call in tb.delete.call_args_list:
+        assert call.kwargs.get("mode") != "all"