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,8 @@
 from contextlib import contextmanager
 from dataclasses import dataclass, field
 import uuid
+
+from google.auth.credentials import AnonymousCredentials
 from mashumaro.helper import pass_through
 
 from functools import lru_cache
@@ -98,6 +100,7 @@ class BigQueryConnectionMethod(StrEnum):
     SERVICE_ACCOUNT = "service-account"
     SERVICE_ACCOUNT_JSON = "service-account-json"
     OAUTH_SECRETS = "oauth-secrets"
+    ANONYMOUS = "anonymous"
 
 
 @dataclass
@@ -126,6 +129,7 @@ class BigQueryCredentials(Credentials):
     schema: Optional[str] = None
     execution_project: Optional[str] = None
     quota_project: Optional[str] = None
+    api_endpoint: Optional[str] = None
     location: Optional[str] = None
     priority: Optional[Priority] = None
     maximum_bytes_billed: Optional[int] = None
@@ -201,6 +205,7 @@ def _connection_keys(self):
             "schema",
             "location",
             "priority",
+            "api_endpoint",
             "maximum_bytes_billed",
             "impersonate_service_account",
             "job_retry_deadline_seconds",
@@ -384,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)
@@ -411,9 +418,12 @@ def get_bigquery_client(cls, profile_credentials):
         execution_project = profile_credentials.execution_project
         quota_project = profile_credentials.quota_project
         location = getattr(profile_credentials, "location", None)
+        api_endpoint = getattr(profile_credentials, "api_endpoint", None)
 
         info = client_info.ClientInfo(user_agent=f"dbt-bigquery-{dbt_version.version}")
-        options = client_options.ClientOptions(quota_project_id=quota_project)
+        options = client_options.ClientOptions(
+            api_endpoint=api_endpoint, quota_project_id=quota_project
+        )
         return google.cloud.bigquery.Client(
             execution_project,
             creds,
@@ -602,8 +612,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_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):
@@ -174,3 +176,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)