fix: cache BigQuery client and surface query errors (#115) (#116)

* fix: cache BigQuery client and surface query errors (#115)

- get_client() now constructs the client once per resource instance and
  reuses it; previously every call re-resolved ADC credentials, and a
  single upsert created three clients
- the location config field is now passed to the client (was silently
  ignored)
- execute_query() no longer swallows all exceptions into an empty
  DataFrame; query failures raise so they surface as asset failures.
  DML/DDL statements (no result schema) still return an empty frame
- fix self-referential class docstring left over from the MotherDuck
  migration
- add unit tests for caching, location passthrough, error surfacing,
  and DML handling

Closes #115

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* style: apply ruff format

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: C00ldudeNoonan

macro_agents/src/macro_agents/defs/resources/bigquery_warehouse.py

@@ -11,7 +11,7 @@
 import polars as pl
 from google.cloud import bigquery
 from google.cloud.exceptions import NotFound
-from pydantic import Field
+from pydantic import Field, PrivateAttr
 
 _IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*){0,2}$")
 
@@ -24,14 +24,21 @@ def _validate_identifier(name: str, kind: str = "identifier") -> str:
 
 
 class BigQueryWarehouseResource(dg.ConfigurableResource):
-    """Dagster resource for BigQuery providing the same public API as BigQueryWarehouseResource."""
+    """Dagster resource for BigQuery.
+
+    Exposes the same public API as the retired MotherDuckResource so assets
+    written against it work unchanged (see the get_connection and
+    drop_create_duck_db_table aliases).
+    """
 
     project: str = Field(description="GCP project ID")
     dataset: str = Field(
         description="Default BigQuery dataset", default="economics_raw"
     )
     location: str = Field(description="BigQuery location", default="US")
 
+    _client: bigquery.Client | None = PrivateAttr(default=None)
+
     def _prepare_google_application_credentials(self) -> None:
         """Support either a credentials file path or inline service-account JSON."""
         credentials = os.getenv("GOOGLE_APPLICATION_CREDENTIALS", "").strip()
@@ -53,9 +60,15 @@ def _prepare_google_application_credentials(self) -> None:
         os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = credentials_path
 
     def get_client(self) -> bigquery.Client:
-        """Return a BigQuery client. On GCE the VM service account is used via ADC."""
-        self._prepare_google_application_credentials()
-        return bigquery.Client(project=self.project)
+        """Return a cached BigQuery client. On GCE the VM service account is used via ADC.
+
+        Client construction resolves ADC credentials, so the client is
+        created once per resource instance and reused across calls.
+        """
+        if self._client is None:
+            self._prepare_google_application_credentials()
+            self._client = bigquery.Client(project=self.project, location=self.location)
+        return self._client
 
     def get_connection(self) -> bigquery.Client:
         """Alias for get_client() — drop-in replacement for MotherDuckResource.get_connection()."""
@@ -167,6 +180,11 @@ def execute_query(
 
         Bare table names are resolved against self.dataset so callers don't
         need to fully qualify every reference.
+
+        Query failures (bad SQL, missing tables, permissions) raise so they
+        surface as asset failures instead of masquerading as empty result
+        sets. DML/DDL statements, which produce no row schema, return an
+        empty DataFrame.
         """
         from google.cloud.bigquery import DatasetReference, QueryJobConfig
 
@@ -175,10 +193,11 @@ def execute_query(
             default_dataset=DatasetReference(self.project, self.dataset)
         )
         job = client.query(query, job_config=job_config)
-        try:
-            return pl.DataFrame(job.result().to_arrow())
-        except Exception:
+        result = job.result()
+        if not result.schema:
+            # DML/DDL statements have no result rows to convert
             return pl.DataFrame()
+        return pl.DataFrame(result.to_arrow())
 
     def table_exists(self, table_name: str) -> bool:
         """Check whether a table exists in BigQuery."""

macro_agents/tests/test_bigquery_warehouse.py

@@ -0,0 +1,88 @@
+"""Tests for BigQueryWarehouseResource client caching and error surfacing (issue #115)."""
+
+from unittest.mock import Mock, patch
+
+import polars as pl
+import pyarrow as pa
+import pytest
+
+from macro_agents.defs.resources.bigquery_warehouse import BigQueryWarehouseResource
+
+
+def _make_resource() -> BigQueryWarehouseResource:
+    return BigQueryWarehouseResource(project="test-project", dataset="test_dataset")
+
+
+class TestGetClientCaching:
+    @patch("macro_agents.defs.resources.bigquery_warehouse.bigquery.Client")
+    def test_client_constructed_once_across_calls(self, mock_client_cls):
+        resource = _make_resource()
+
+        first = resource.get_client()
+        second = resource.get_client()
+
+        assert first is second
+        mock_client_cls.assert_called_once_with(project="test-project", location="US")
+
+    @patch("macro_agents.defs.resources.bigquery_warehouse.bigquery.Client")
+    def test_location_config_is_passed_through(self, mock_client_cls):
+        resource = BigQueryWarehouseResource(
+            project="test-project", dataset="test_dataset", location="EU"
+        )
+
+        resource.get_client()
+
+        mock_client_cls.assert_called_once_with(project="test-project", location="EU")
+
+    @patch("macro_agents.defs.resources.bigquery_warehouse.bigquery.Client")
+    def test_get_connection_reuses_cached_client(self, mock_client_cls):
+        resource = _make_resource()
+
+        client = resource.get_client()
+        connection = resource.get_connection()
+
+        assert connection is client
+        mock_client_cls.assert_called_once()
+
+
+class TestExecuteQueryErrorSurfacing:
+    @patch("macro_agents.defs.resources.bigquery_warehouse.bigquery.Client")
+    def test_query_failure_raises_instead_of_returning_empty(self, mock_client_cls):
+        mock_job = Mock()
+        mock_job.result.side_effect = RuntimeError("Table not found: missing_table")
+        mock_client_cls.return_value.query.return_value = mock_job
+        resource = _make_resource()
+
+        with pytest.raises(RuntimeError, match="missing_table"):
+            resource.execute_query("SELECT * FROM missing_table")
+
+    @patch("macro_agents.defs.resources.bigquery_warehouse.bigquery.Client")
+    def test_select_returns_polars_dataframe(self, mock_client_cls):
+        arrow_table = pa.table({"value": [1, 2, 3]})
+        mock_result = Mock()
+        mock_result.schema = ["value"]
+        mock_result.to_arrow.return_value = arrow_table
+        mock_job = Mock()
+        mock_job.result.return_value = mock_result
+        mock_client_cls.return_value.query.return_value = mock_job
+        resource = _make_resource()
+
+        df = resource.execute_query("SELECT value FROM some_table")
+
+        assert isinstance(df, pl.DataFrame)
+        assert df["value"].to_list() == [1, 2, 3]
+
+    @patch("macro_agents.defs.resources.bigquery_warehouse.bigquery.Client")
+    def test_dml_statement_returns_empty_dataframe(self, mock_client_cls):
+        mock_result = Mock()
+        mock_result.schema = []
+        mock_job = Mock()
+        mock_job.result.return_value = mock_result
+        mock_client_cls.return_value.query.return_value = mock_job
+        resource = _make_resource()
+
+        df = resource.execute_query("UPDATE some_table SET x = 1 WHERE TRUE")
+
+        assert isinstance(df, pl.DataFrame)
+        assert df.is_empty()
+        mock_result.to_arrow.assert_not_called()