feat(dataconnect): Implementation and Testing of the first half of _DataConnectApiClient class (#965)

*Core API Client Helpers*

- Implemented `_validate_inputs` to enforce strict validation on query structures, options, variables, and custom impersonation claims.

- Implemented `_prepare_graphql_payload` to construct the GraphQL JSON payload, seamlessly serializing variables (including nested dataclasses) and optional extensions.

- Implemented `_get_firebase_dataconnect_service_url` to dynamically build endpoint URLs for both production and local emulator hosts.

- Implemented `_get_headers` to inject standard telemetry metrics (X-Firebase-Client and x-goog-api-client).

*Validation & Serialization Refinements*

- Enforced that GraphQL variables must be either a Mapping (like standard dicts) or a dataclass.
- Added support for runtime validation of subscripted generic types (e.g., Dict, Mapping) using typing.get_origin().
- Refactored `Impersonation` to inherit from `dict`, resolving type-checking mismatches while maintaining runtime dictionary behavior. Extracted variable and impersonation validation into dedicated helper methods.

*Testing & Code Health*

- Added comprehensive unit test coverage for the client constructor, input validation, headers, and payload serialization (using realistic nested dataclasses).

Author: mk2023

firebase_admin/_utils.py

@@ -15,6 +15,8 @@
 """Internal utilities common to all modules."""
 
 import json
+import os
+import re
 from platform import python_version
 from typing import Callable, Optional
 
@@ -345,3 +347,16 @@ def __init__(self):
 
     def refresh(self, request):
         pass
+
+
+def get_emulator_host(env_var_name: str) -> Optional[str]:
+    """Retrieves and validates the host from the specified emulator environment variable."""
+    emulator_host = os.environ.get(env_var_name)
+    if emulator_host:
+        if not re.match(r'^(?:\[[a-fA-F0-9:]+\]|[a-zA-Z0-9._-]+):[0-9]+$', emulator_host):
+            raise ValueError(
+                f'Invalid {env_var_name}: "{emulator_host}". It must follow format '
+                '"host:port".'
+            )
+        return emulator_host
+    return None

firebase_admin/dataconnect.py

@@ -18,14 +18,39 @@
 Firebase apps.
 """
 
-from dataclasses import dataclass
-from typing import Dict, Optional
+from collections.abc import Mapping
+from dataclasses import dataclass, asdict, is_dataclass
+import typing
+from typing import Any, Dict, Generic, Optional, Type, TypeVar, Union
+import firebase_admin
+from firebase_admin import _utils, _http_client, App
 
-from firebase_admin import _utils, App
-
-__all__ = ['ConnectorConfig', 'DataConnect', 'client']
+__all__ = [
+    'ConnectorConfig',
+    'DataConnect',
+    'client',
+    'GraphqlOptions',
+    'Impersonation',
+    'ExecuteGraphqlResponse',
+]
 
 _DATA_CONNECT_ATTRIBUTE = '_data_connect'
+_DATA_CONNECT_PROD_URL = 'https://firebasedataconnect.googleapis.com'
+_API_VERSION = 'v1'
+
+_SERVICES_URL_FORMAT = (
+    '{host}/{version}/projects/{project_id}/locations/{location_id}'
+    '/services/{service_id}:{endpoint_id}'
+)
+
+_EMULATOR_SERVICES_URL_FORMAT = (
+    'http://{host}/{version}/projects/{project_id}/locations/{location_id}'
+    '/services/{service_id}:{endpoint_id}'
+)
+
+# Generic Type Parameters
+_Data = TypeVar("_Data")
+_Variables = TypeVar("_Variables")
 
 @dataclass(frozen=True)
 class ConnectorConfig:
@@ -122,3 +147,190 @@ def client(config: ConnectorConfig, app: Optional[App] = None) -> DataConnect:
     dc_service = _utils.get_app_service(app, _DATA_CONNECT_ATTRIBUTE, _DataConnectService)
 
     return dc_service.get_client(config)
+
+
+
+class Impersonation(dict):
+    """Represents impersonation configuration for DataConnect requests."""
+
+    @staticmethod
+    def unauthenticated() -> 'Impersonation':
+        """Returns impersonation configuration for unauthenticated requests."""
+        return Impersonation(unauthenticated=True)
+
+    @staticmethod
+    def authenticated(auth_claims: Dict[str, Any]) -> 'Impersonation':
+        """Returns impersonation configuration for authenticated requests.
+
+        # TODO: More strongly type auth_claims later.
+        """
+        return Impersonation(authClaims=auth_claims)
+
+
+@dataclass
+class GraphqlOptions(Generic[_Variables]):
+    variables: Optional[_Variables] = None
+    operation_name: Optional[str] = None
+    impersonate: Optional[Union[Impersonation, Dict[str, Any]]] = None
+
+
+@dataclass
+class ExecuteGraphqlResponse(Generic[_Data]):
+    data: _Data
+
+
+def _get_emulator_host() -> Optional[str]:
+    return _utils.get_emulator_host("DATA_CONNECT_EMULATOR_HOST")
+
+
+class _DataConnectApiClient:
+    """Internal client for sending requests to the Firebase Data Connect backend.
+
+    Attributes:
+        connector_config: The connector configuration specifying the service,
+            location, and connector name.
+        app: The Firebase App instance associated with this client.
+    """
+
+    def __init__(self, connector_config: ConnectorConfig, app: App) -> None:
+        if not isinstance(app, App):
+            raise ValueError(
+                'Second argument passed to DataConnectApiClient must be a valid '
+                'Firebase app instance.'
+            )
+        self._connector_config = connector_config
+        self._app = app
+
+        self._project_id = app.project_id
+        if not self._project_id:
+            raise ValueError(
+                'Failed to determine project ID. Initialize the SDK with service '
+                'account credentials or set project ID as an app option. Alternatively, set the '
+                'GOOGLE_CLOUD_PROJECT environment variable.')
+
+        self._emulator_host = _get_emulator_host()
+        if self._emulator_host:
+            self._credential = _utils.EmulatorAdminCredentials()
+        else:
+            self._credential = app.credential.get_credential()
+
+        self._http_client = _http_client.JsonHttpClient(credential=self._credential)
+
+    def _validate_variables_type(
+        self,
+        variables: Any,
+        variable_type: Optional[Type[Any]] = None
+    ) -> None:
+        """Validates variables against expected type."""
+        if variables is not None:
+            if not (isinstance(variables, Mapping) or is_dataclass(variables)):
+                raise ValueError("variables must be a collections.abc.Mapping or a dataclass")
+            if variable_type is not None:
+                expected_type = typing.get_origin(variable_type) or variable_type
+                if not isinstance(variables, expected_type):
+                    type_name = getattr(expected_type, '__name__', str(expected_type))
+                    raise ValueError(f"variables must be of type {type_name}")
+
+    def _validate_impersonation_options(self, impersonate: Any) -> None:
+        """Validates impersonation dictionary options."""
+        if impersonate is not None:
+            if not isinstance(impersonate, dict):
+                raise ValueError('impersonate option must be a dictionary')
+            if 'unauthenticated' not in impersonate and 'authClaims' not in impersonate:
+                raise ValueError(
+                    "impersonate option must contain either "
+                    "'unauthenticated' or 'authClaims'"
+                )
+            if 'unauthenticated' in impersonate and 'authClaims' in impersonate:
+                raise ValueError(
+                    "impersonate option cannot contain both "
+                    "'unauthenticated' and 'authClaims'"
+                )
+            if 'unauthenticated' in impersonate:
+                if not isinstance(impersonate['unauthenticated'], bool):
+                    raise ValueError("'unauthenticated' claim must be a boolean")
+            if 'authClaims' in impersonate:
+                if not isinstance(impersonate['authClaims'], dict):
+                    raise ValueError("'authClaims' claim must be a dictionary")
+
+    def _validate_graphql_options(
+        self,
+        graphql_options: Optional[GraphqlOptions[Any]],
+        variable_type: Optional[Type[Any]] = None
+    ) -> None:
+        """Validates GraphqlOptions inputs at runtime."""
+        if graphql_options is not None:
+            if not isinstance(graphql_options, GraphqlOptions):
+                raise ValueError('options must be a GraphqlOptions instance')
+
+            # Validate Variables against expected variable_type
+            self._validate_variables_type(graphql_options.variables, variable_type)
+
+            # Validate Operation Name (if it exists)
+            operation_name = graphql_options.operation_name
+            if operation_name is not None:
+                if not isinstance(operation_name, str):
+                    raise ValueError('operation_name must be a string')
+                if not operation_name.strip():
+                    raise ValueError('operation_name must be a non-empty string')
+
+            # Validate Impersonation (if it exists)
+            self._validate_impersonation_options(graphql_options.impersonate)
+
+    def _prepare_graphql_payload(
+        self,
+        graphql_query: str,
+        graphql_options: Optional[GraphqlOptions[_Variables]]
+    ) -> Dict[str, Any]:
+        """Serializes input query and options to JSON-compatible dictionary."""
+        payload = {
+            "query": graphql_query
+        }
+
+        if graphql_options is not None:
+            if graphql_options.variables is not None:
+                if is_dataclass(graphql_options.variables):
+                    payload["variables"] = asdict(graphql_options.variables)
+                else:
+                    payload["variables"] = graphql_options.variables
+
+            if graphql_options.operation_name is not None:
+                payload["operationName"] = graphql_options.operation_name.strip()
+
+            if graphql_options.impersonate is not None:
+                payload["extensions"] = {
+                    "impersonate": graphql_options.impersonate
+                }
+
+        return payload
+
+    def _get_firebase_dataconnect_service_url(self, method_name: str) -> str:
+        """Build and return the URL for a Firebase Data Connect API method."""
+        project_id = self._project_id
+        location = self._connector_config.location
+        service_id = self._connector_config.service_id
+
+        if self._emulator_host:
+            return _EMULATOR_SERVICES_URL_FORMAT.format(
+                host=self._emulator_host,
+                version=_API_VERSION,
+                project_id=project_id,
+                location_id=location,
+                service_id=service_id,
+                endpoint_id=method_name
+            )
+        return _SERVICES_URL_FORMAT.format(
+            host=_DATA_CONNECT_PROD_URL,
+            version=_API_VERSION,
+            project_id=project_id,
+            location_id=location,
+            service_id=service_id,
+            endpoint_id=method_name
+        )
+
+    def _get_headers(self) -> Dict[str, str]:
+        """Build and return the headers for a Firebase Data Connect API call."""
+        return {
+            "X-Firebase-Client": f"fire-admin-python/{firebase_admin.__version__}",
+            "x-goog-api-client": _utils.get_metrics_header(),
+        }

firebase_admin/functions.py

@@ -18,7 +18,6 @@
 from datetime import datetime, timedelta, timezone
 from urllib import parse
 import re
-import os
 import json
 from base64 import b64encode
 from typing import Any, Optional, Dict
@@ -62,14 +61,7 @@
 _DEFAULT_LOCATION = 'us-central1'
 
 def _get_emulator_host() -> Optional[str]:
-    emulator_host = os.environ.get(_EMULATOR_HOST_ENV_VAR)
-    if emulator_host:
-        if '//' in emulator_host:
-            raise ValueError(
-                f'Invalid {_EMULATOR_HOST_ENV_VAR}: "{emulator_host}". It must follow format '
-                '"host:port".')
-        return emulator_host
-    return None
+    return _utils.get_emulator_host(_EMULATOR_HOST_ENV_VAR)
 
 
 def _get_functions_service(app) -> _FunctionsService: