diff --git a/firebase_admin/dataconnect.py b/firebase_admin/dataconnect.py index 201e3b12..706106c2 100644 --- a/firebase_admin/dataconnect.py +++ b/firebase_admin/dataconnect.py @@ -18,14 +18,39 @@ Firebase apps. """ -from dataclasses import dataclass -from typing import Dict, Optional +import os +from dataclasses import dataclass, asdict, is_dataclass +from typing import Any, Dict, Generic, Optional, TypeVar +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 +_T = TypeVar("_T") +_V = TypeVar("_V") @dataclass(frozen=True) class ConnectorConfig: @@ -122,3 +147,178 @@ 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: + """Represents impersonation configuration for DataConnect requests.""" + + @staticmethod + def unauthenticated() -> Dict[str, bool]: + """Returns impersonation configuration for unauthenticated requests.""" + return {"unauthenticated": True} + + @staticmethod + def authenticated(auth_claims: Dict[str, Any]) -> Dict[str, Any]: + """Returns impersonation configuration for authenticated requests.""" + return {"authClaims": auth_claims} + + +@dataclass +class GraphqlOptions(Generic[_V]): + variables: Optional[_V] = None + operation_name: Optional[str] = None + impersonate: Optional[Impersonation] = None + + +@dataclass +class ExecuteGraphqlResponse(Generic[_T]): + data: _T + + +def _get_emulator_host() -> Optional[str]: + emulator_host = os.environ.get("DATA_CONNECT_EMULATOR_HOST") + if emulator_host: + if "//" in emulator_host: + raise ValueError( + f'Invalid DATA_CONNECT_EMULATOR_HOST: "{emulator_host}". It must follow format ' + '"host:port".' + ) + return emulator_host + return None + + +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( + 'First 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_inputs( + self, + graphql_query: str, + graphql_options: Optional[GraphqlOptions[Any]], + variable_type: Any = None + ) -> None: + """Validates query and GraphqlOptions inputs at runtime.""" + # Validate the Query + if not isinstance(graphql_query, str) or not graphql_query.strip(): + raise ValueError('query must be a non-empty string') + + # Validate Options (if they exist) + 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 + variables = graphql_options.variables + if variables is not None and variable_type is not None: + if not isinstance(variables, variable_type): + raise ValueError(f"variables must be of type {variable_type.__name__}") + + # Validate Operation Name (if it exists) + operation_name = graphql_options.operation_name + if operation_name is not None: + if not isinstance(operation_name, str) or not operation_name.strip(): + raise ValueError('operation_name must be a non-empty string') + + # Validate Impersonation (if it exists) + impersonate = graphql_options.impersonate + 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: + 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 _prepare_graphql_payload( + self, + graphql_query: str, + graphql_options: Optional[GraphqlOptions[Any]] + ) -> 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 + + 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(), + } diff --git a/tests/test_data_connect.py b/tests/test_data_connect.py index c226b12b..f6b7a390 100644 --- a/tests/test_data_connect.py +++ b/tests/test_data_connect.py @@ -15,6 +15,8 @@ """Test cases for the firebase_admin.dataconnect module.""" from unittest import mock +from dataclasses import dataclass +from google.auth import credentials as google_auth_credentials import pytest import firebase_admin @@ -331,3 +333,336 @@ def test_overall_client_retrieval_and_caching(self): assert client1_app2.app is self.app2 assert client1_app2.config is self.config1 assert client1_app2 is not client1a + + +class TestDataConnectApiClientConstructor: + + def setup_method(self): + self.cred = testutils.MockCredential() + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_constructor_invalid_app(self): + msg = ( + "First argument passed to DataConnectApiClient must be a valid " + "Firebase app instance." + ) + with pytest.raises(ValueError, match=msg): + dataconnect._DataConnectApiClient(BASE_CONFIG, None) + + def test_constructor_missing_project_id(self): + class CredentialWithoutProjectId(firebase_admin.credentials.Base): + def get_credential(self): + class DummyGoogleCred(google_auth_credentials.Credentials): + def refresh(self, request): + pass + return DummyGoogleCred() + + app_no_project_id = firebase_admin.initialize_app( + CredentialWithoutProjectId(), + name="no-project-id-app" + ) + try: + with pytest.raises(ValueError, match="Failed to determine project ID"): + dataconnect._DataConnectApiClient(BASE_CONFIG, app_no_project_id) + finally: + firebase_admin.delete_app(app_no_project_id) + + def test_constructor_connector_config(self): + app = firebase_admin.initialize_app(self.cred, options={'projectId': 'test-project'}) + api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, app) + assert api_client._connector_config is BASE_CONFIG + + +class TestDataConnectApiClientValidateInputs: + + def setup_method(self): + self.cred = testutils.MockCredential() + self.app = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'} + ) + self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app) + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_validate_inputs_valid(self): + # Valid query, no options + self.api_client._validate_inputs("query { hello }", None) + + # Valid query, valid options + options = dataconnect.GraphqlOptions(variables={"foo": "bar"}) + self.api_client._validate_inputs("query { hello }", options) + + def test_validate_inputs_valid_impersonate(self): + # Valid unauthenticated impersonation + imp_unauth = dataconnect.Impersonation.unauthenticated() + options = dataconnect.GraphqlOptions(impersonate=imp_unauth) + self.api_client._validate_inputs("query { hello }", options) + + # Valid authenticated impersonation + imp_auth = dataconnect.Impersonation.authenticated( + {"sub": "authenticated-UUID"} + ) + options = dataconnect.GraphqlOptions(impersonate=imp_auth) + self.api_client._validate_inputs("query { hello }", options) + + + def test_validate_inputs_valid_variables(self): + @dataclass + class User: + user_id: str + name: str + address: str + + @dataclass + class UsersResponse: + users: list[User] + + users_val = [User(user_id="1", name="Fred", address="123 Road")] + valid_variables = UsersResponse(users=users_val) + options = dataconnect.GraphqlOptions(variables=valid_variables) + self.api_client._validate_inputs("query { hello }", options, UsersResponse) + + def test_validate_inputs_invalid_query_type(self): + + with pytest.raises(ValueError, match="query must be a non-empty string"): + self.api_client._validate_inputs(None, None) + with pytest.raises(ValueError, match="query must be a non-empty string"): + self.api_client._validate_inputs(123, None) + + def test_validate_inputs_empty_query(self): + with pytest.raises(ValueError, match="query must be a non-empty string"): + self.api_client._validate_inputs("", None) + with pytest.raises(ValueError, match="query must be a non-empty string"): + self.api_client._validate_inputs(" ", None) + + def test_validate_inputs_invalid_options(self): + with pytest.raises(ValueError, match="options must be a GraphqlOptions instance"): + self.api_client._validate_inputs("query { hello }", "invalid-options") + + def test_validate_inputs_invalid_impersonate(self): + # impersonate must be dict + options = dataconnect.GraphqlOptions(impersonate="invalid") + with pytest.raises(ValueError, match="impersonate option must be a dictionary"): + self.api_client._validate_inputs("query { hello }", options) + + # impersonate must have either unauthenticated or authClaims + options = dataconnect.GraphqlOptions(impersonate={"invalid_key": True}) + msg = ( + "impersonate option must contain either " + "'unauthenticated' or 'authClaims'" + ) + with pytest.raises(ValueError, match=msg): + self.api_client._validate_inputs("query { hello }", options) + + # unauthenticated must be boolean + options = dataconnect.GraphqlOptions(impersonate={"unauthenticated": "not-bool"}) + with pytest.raises(ValueError, match="'unauthenticated' claim must be a boolean"): + self.api_client._validate_inputs("query { hello }", options) + + # authClaims must be a dict + options = dataconnect.GraphqlOptions(impersonate={"authClaims": "not-dict"}) + with pytest.raises(ValueError, match="'authClaims' claim must be a dictionary"): + self.api_client._validate_inputs("query { hello }", options) + + def test_validate_inputs_invalid_operation_name(self): + options = dataconnect.GraphqlOptions(operation_name=123) + with pytest.raises(ValueError, match="operation_name must be a non-empty string"): + self.api_client._validate_inputs("query { hello }", options) + options = dataconnect.GraphqlOptions(operation_name="") + with pytest.raises(ValueError, match="operation_name must be a non-empty string"): + self.api_client._validate_inputs("query { hello }", options) + + def test_validate_inputs_invalid_variables(self): + @dataclass + class User: + user_id: str + name: str + address: str + + @dataclass + class UsersResponse: + users: list[User] + + options = dataconnect.GraphqlOptions(variables="not-users-response") + with pytest.raises(ValueError, match="variables must be of type UsersResponse"): + self.api_client._validate_inputs("query { hello }", options, UsersResponse) + + +class TestDataConnectApiClientPrepareGraphqlPayload: + + def setup_method(self): + self.cred = testutils.MockCredential() + self.app = firebase_admin.initialize_app(self.cred, options={'projectId': 'test-project'}) + self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app) + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_prepare_graphql_payload_only_query(self): + payload = self.api_client._prepare_graphql_payload("query { hello }", None) + assert payload == {"query": "query { hello }"} + + def test_prepare_graphql_payload_with_variables(self): + options = dataconnect.GraphqlOptions(variables={"foo": "bar"}) + payload = self.api_client._prepare_graphql_payload("query { hello }", options) + assert payload == { + "query": "query { hello }", + "variables": {"foo": "bar"} + } + + def test_prepare_graphql_payload_with_dataclass_variables(self): + @dataclass + class User: + user_id: str + name: str + address: str + + @dataclass + class UsersResponse: + users: list[User] + + users_val = [User(user_id="1", name="Fred", address="123 Road")] + valid_variables = UsersResponse(users=users_val) + options = dataconnect.GraphqlOptions(variables=valid_variables) + payload = self.api_client._prepare_graphql_payload("query { hello }", options) + assert payload == { + "query": "query { hello }", + "variables": { + "users": [ + { + "user_id": "1", + "name": "Fred", + "address": "123 Road" + } + ] + } + } + + def test_prepare_graphql_payload_with_operation_name(self): + options = dataconnect.GraphqlOptions(operation_name="myOp") + payload = self.api_client._prepare_graphql_payload("query { hello }", options) + assert payload == { + "query": "query { hello }", + "operationName": "myOp" + } + + def test_prepare_graphql_payload_with_impersonate_unauthenticated(self): + imp_unauth = dataconnect.Impersonation.unauthenticated() + options = dataconnect.GraphqlOptions(impersonate=imp_unauth) + payload = self.api_client._prepare_graphql_payload("query { hello }", options) + assert payload == { + "query": "query { hello }", + "extensions": { + "impersonate": {"unauthenticated": True} + } + } + + def test_prepare_graphql_payload_with_impersonate_authenticated(self): + imp_auth = dataconnect.Impersonation.authenticated( + {"sub": "authenticated-UUID"} + ) + options = dataconnect.GraphqlOptions(impersonate=imp_auth) + payload = self.api_client._prepare_graphql_payload("query { hello }", options) + assert payload == { + "query": "query { hello }", + "extensions": { + "impersonate": {"authClaims": {"sub": "authenticated-UUID"}} + } + } + + def test_prepare_graphql_payload_with_all_fields(self): + @dataclass + class User: + user_id: str + name: str + address: str + + @dataclass + class UsersResponse: + users: list[User] + + users_val = [User(user_id="1", name="Fred", address="123 Road")] + valid_variables = UsersResponse(users=users_val) + imp_auth = dataconnect.Impersonation.authenticated( + {"sub": "authenticated-UUID"} + ) + options = dataconnect.GraphqlOptions( + variables=valid_variables, + operation_name="getUsers", + impersonate=imp_auth + ) + payload = self.api_client._prepare_graphql_payload("query { hello }", options) + assert payload == { + "query": "query { hello }", + "operationName": "getUsers", + "variables": { + "users": [ + { + "user_id": "1", + "name": "Fred", + "address": "123 Road" + } + ] + }, + "extensions": { + "impersonate": {"authClaims": {"sub": "authenticated-UUID"}} + } + } + + +class TestDataConnectApiClientServiceUrl: + + def setup_method(self): + self.cred = testutils.MockCredential() + self.app = firebase_admin.initialize_app(self.cred, options={'projectId': 'test-project'}) + self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app) + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_get_firebase_dataconnect_service_url_production(self): + url = self.api_client._get_firebase_dataconnect_service_url("executeGraphql") + expected = ( + "https://firebasedataconnect.googleapis.com/v1" + "/projects/test-project/locations/us-east4" + "/services/starterproject:executeGraphql" + ) + assert url == expected + + def test_get_firebase_dataconnect_service_url_emulator(self, monkeypatch): + monkeypatch.setenv("DATA_CONNECT_EMULATOR_HOST", "localhost:9399") + api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app) + url = api_client._get_firebase_dataconnect_service_url("executeGraphql") + expected = ( + "http://localhost:9399/v1" + "/projects/test-project/locations/us-east4" + "/services/starterproject:executeGraphql" + ) + assert url == expected + + +class TestDataConnectApiClientGetHeaders: + + def setup_method(self): + self.cred = testutils.MockCredential() + self.app = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'} + ) + self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app) + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_get_headers(self): + headers = self.api_client._get_headers() + assert isinstance(headers, dict) + assert headers.get("X-Firebase-Client") == f"fire-admin-python/{firebase_admin.__version__}" + assert headers.get("x-goog-api-client") == _utils.get_metrics_header()