FEATURE: Add support for google.auth.credentials.AnonymousCredentials

.changes/unreleased/Features-20241020-112955.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support google.auth.credentials.AnonymousCredentials for bigquery enabling to unit test models in isolation locally.
+time: 2024-10-20T11:29:55.098322635Z
+custom:
+    Author: shrivastava-ankur
+    Issue: "1377"

dbt/adapters/bigquery/connections.py

@@ -5,6 +5,9 @@
 from contextlib import contextmanager
 from dataclasses import dataclass, field
 import uuid
+
+from google.api_core.client_options import ClientOptions
+from google.auth.credentials import AnonymousCredentials
 from mashumaro.helper import pass_through
 
 from functools import lru_cache
@@ -98,6 +101,7 @@ class BigQueryConnectionMethod(StrEnum):
     SERVICE_ACCOUNT = "service-account"
     SERVICE_ACCOUNT_JSON = "service-account-json"
     OAUTH_SECRETS = "oauth-secrets"
+    ANONYMOUS = "anonymous"
 
 
 @dataclass
@@ -134,6 +138,7 @@ class BigQueryCredentials(Credentials):
     job_retries: Optional[int] = 1
     job_creation_timeout_seconds: Optional[int] = None
     job_execution_timeout_seconds: Optional[int] = None
+    client_options: Optional[Dict[str, Any]] = None
 
     # Keyfile json creds (unicode or base 64 encoded)
     keyfile: Optional[str] = None
@@ -206,6 +211,7 @@ def _connection_keys(self):
             "job_retries",
             "job_creation_timeout_seconds",
             "job_execution_timeout_seconds",
+            "client_options",
             "timeout_seconds",
             "client_id",
             "token_uri",
@@ -383,6 +389,8 @@ def get_google_credentials(cls, profile_credentials) -> GoogleCredentials:
                 token_uri=profile_credentials.token_uri,
                 scopes=profile_credentials.scopes,
             )
+        elif method == BigQueryConnectionMethod.ANONYMOUS:
+            return AnonymousCredentials()
 
         error = 'Invalid `method` in profile: "{}"'.format(method)
         raise FailedToConnectError(error)
@@ -409,13 +417,16 @@ def get_bigquery_client(cls, profile_credentials):
         creds = cls.get_credentials(profile_credentials)
         execution_project = profile_credentials.execution_project
         location = getattr(profile_credentials, "location", None)
+        client_options_kwargs = getattr(profile_credentials, "client_options", None)
 
+        options = ClientOptions(**client_options_kwargs) if client_options_kwargs else None
         info = client_info.ClientInfo(user_agent=f"dbt-bigquery-{dbt_version.version}")
         return google.cloud.bigquery.Client(
             execution_project,
             creds,
             location=location,
             client_info=info,
+            client_options=options,
         )
 
     @classmethod
@@ -598,8 +609,11 @@ def execute(
             conn = self.get_thread_connection()
             client = conn.handle
             # use anonymous table for num_rows
-            query_table = client.get_table(query_job.destination)
-            num_rows = query_table.num_rows
+            if query_job.destination:
+                query_table = client.get_table(query_job.destination)
+                num_rows = query_table.num_rows
+            else:
+                num_rows = None
 
         # set common attributes
         bytes_processed = query_job.total_bytes_processed

tests/unit/test_bigquery_adapter.py

@@ -403,6 +403,7 @@ def test_location_user_agent(self, mock_bq, mock_auth_default):
             creds,
             location="Luna Station",
             client_info=HasUserAgent(),
+            client_options=None,
         )
 
 

tests/unit/test_bigquery_connection_manager.py

@@ -1,14 +1,16 @@
 import json
 import unittest
 from contextlib import contextmanager
+
+from google.auth.credentials import AnonymousCredentials
 from requests.exceptions import ConnectionError
 from unittest.mock import patch, MagicMock, Mock, ANY
 
 import dbt.adapters
 
 from dbt.adapters.bigquery import BigQueryCredentials
 from dbt.adapters.bigquery import BigQueryRelation
-from dbt.adapters.bigquery.connections import BigQueryConnectionManager
+from dbt.adapters.bigquery.connections import BigQueryConnectionManager, BigQueryConnectionMethod
 
 
 class TestBigQueryConnectionManager(unittest.TestCase):
@@ -172,3 +174,17 @@ def _copy_table(self, write_disposition):
             database="project", schema="dataset", identifier="table2"
         )
         self.connections.copy_bq_table(source, destination, write_disposition)
+
+    def test_local_test_container_connection(self):
+        # Create a BigQueryCredentials instance with the anonymous method
+        credentials = BigQueryCredentials(
+            method=BigQueryConnectionMethod.ANONYMOUS,
+            database="test-database",
+            schema="test-schema",
+        )
+
+        # Get the Google credentials using the connection manager
+        google_credentials = BigQueryConnectionManager.get_google_credentials(credentials)
+
+        # Assert that the credentials are an instance of AnonymousCredentials
+        self.assertIsInstance(google_credentials, AnonymousCredentials)