fix(bigtable): avoid DuplicateCredentialArgs in BigtableMetricsExporter

packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import copy
 import logging
 import time
 from collections import defaultdict
@@ -167,6 +168,27 @@ class BigtableMetricsExporter(MetricExporter):
 
     def __init__(self, *client_args, **client_kwargs):
         super().__init__()
+        if client_kwargs.get("credentials") is not None:
+            client_options = client_kwargs.get("client_options")
+            if isinstance(client_options, dict):
+                if (
+                    client_options.get("credentials_file") is not None
+                    or client_options.get("api_key") is not None
+                ):
+                    client_options = {
+                        k: v
+                        for k, v in client_options.items()
+                        if k not in ("credentials_file", "api_key")
+                    }
+                    client_kwargs["client_options"] = client_options
+            elif client_options is not None and (
+                getattr(client_options, "credentials_file", None) is not None
+                or getattr(client_options, "api_key", None) is not None
+            ):
+                client_options = copy.copy(client_options)
+                client_options.credentials_file = None
+                client_options.api_key = None
+                client_kwargs["client_options"] = client_options
         self.client = MetricServiceClient(*client_args, **client_kwargs)
         self.prefix = "bigtable.googleapis.com/internal/client"
 

packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py

@@ -176,6 +176,47 @@ def test_ctor_mocks(self):
             assert instance.prefix == "bigtable.googleapis.com/internal/client"
             mock_client.assert_called_once_with(*args, **kwargs)
 
+    def test_ctor_with_credentials_and_credentials_file(self):
+        """Passing both resolved credentials and a ClientOptions with
+        credentials_file (or api_key) should not raise DuplicateCredentialArgs
+        or mutate the caller's ClientOptions."""
+        import google.auth.credentials
+        from google.api_core.client_options import ClientOptions
+
+        class _FakeScopedCreds(
+            google.auth.credentials.Scoped, google.auth.credentials.Credentials
+        ):
+            @property
+            def requires_scopes(self):
+                return False
+
+            def with_scopes(self, scopes, default_scopes=None):
+                return self
+
+            def refresh(self, request):
+                pass
+
+        creds = _FakeScopedCreds()
+        opts = ClientOptions(
+            credentials_file="/fake/sa_key.json",
+            api_endpoint="monitoring.googleapis.com",
+        )
+        exporter = BigtableMetricsExporter(credentials=creds, client_options=opts)
+        assert exporter.client is not None
+        assert opts.credentials_file == "/fake/sa_key.json"
+
+        dict_opts = {
+            "credentials_file": "/fake/sa_key.json",
+            "api_key": "fake-key",
+            "api_endpoint": "monitoring.googleapis.com",
+        }
+        exporter_from_dict = BigtableMetricsExporter(
+            credentials=creds, client_options=dict_opts
+        )
+        assert exporter_from_dict.client is not None
+        assert dict_opts["credentials_file"] == "/fake/sa_key.json"
+        assert dict_opts["api_key"] == "fake-key"
+
     @pytest.mark.parametrize(
         "value,expected_field",
         [