feat(fdc): Add execute_graphql signatures and integration test suite

firebase_admin/dataconnect.py

@@ -53,6 +53,9 @@
     '/services/{service_id}:{endpoint_id}'
 )
 
+_EXECUTE_GRAPHQL_ENDPOINT = 'executeGraphql'
+_EXECUTE_GRAPHQL_READ_ENDPOINT = 'executeGraphqlRead'
+
 # Generic Type Parameters
 _Data = TypeVar("_Data")
 _Variables = TypeVar("_Variables")
@@ -101,7 +104,76 @@ def __post_init__(self):
             raise ValueError("connector cannot be empty")
 
 
+class Impersonation(dict):
+    """Represents impersonation configuration for DataConnect requests.
+
+    It is recommended to construct instances using the static factory methods
+    :meth:`unauthenticated` or :meth:`authenticated`.
+    """
+
+    def __init__(
+        self,
+        *,
+        unauthenticated: Optional[bool] = None,
+        auth_claims: Optional[Dict[str, Any]] = None,
+        authClaims: Optional[Dict[str, Any]] = None
+    ) -> None:
+        if auth_claims is not None and authClaims is not None:
+            raise ValueError("Cannot specify both 'auth_claims' and 'authClaims'.")
+
+        claims = auth_claims if auth_claims is not None else authClaims
+
+        if unauthenticated is None and claims is None:
+            raise ValueError(
+                "Impersonation requires either 'unauthenticated=True' or 'auth_claims'."
+            )
+        if unauthenticated is not None and claims is not None:
+            raise ValueError("Cannot specify both 'unauthenticated' and 'auth_claims'.")
+
+        if unauthenticated is not None:
+            if not isinstance(unauthenticated, bool):
+                raise ValueError("'unauthenticated' must be a boolean.")
+            super().__init__(unauthenticated=unauthenticated)
+        else:
+            if not isinstance(claims, dict):
+                raise ValueError("'auth_claims' must be a dictionary.")
+            super().__init__(authClaims=claims)
+
+    @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(auth_claims=auth_claims)
+
+
+@dataclass
+class GraphqlOptions(Generic[_Variables]):
+    variables: Optional[_Variables] = None
+    operation_name: Optional[str] = None
+    impersonate: Optional[Union[Impersonation, Dict[str, Any]]] = None
+
+
+# TODO(b/406281627): Add support for partial errors.
+@dataclass
+class ExecuteGraphqlResponse(Generic[_Data]):
+    """Represents the response from a DataConnect GraphQL execution.
+
+    Attributes:
+        data: The raw JSON dictionary returned by the GraphQL execution.
+    """
+    data: _Data
+
+
 class DataConnect:
+
+
     """Represents a Firebase Data Connect client instance. 
 
        This client provides access to the Firebase Data Connect service 
@@ -126,6 +198,66 @@ def app(self) -> App:
     def config(self) -> ConnectorConfig:
         return self._config
 
+    def execute_graphql(
+        self,
+        query: str,
+        options: Optional[GraphqlOptions[_Variables]] = None,
+        variables_type: Type[_Variables] = Any,
+    ) -> ExecuteGraphqlResponse[Any]:
+        """Executes a GraphQL query or mutation and returns the result.
+
+        Args:
+            query: string containing the GraphQL query
+            options: GraphqlOptions instance containing operational parameters such as
+                variables, operation name, or impersonation context (optional).
+            variables_type: The expected structure for the request variables
+
+        Returns:
+            ExecuteGraphqlResponse: An ExecuteGraphqlResponse containing the raw
+                response data dictionary.
+
+        Raises:
+            ValueError: If the arguments are invalid from the local inputs side.
+            InvalidArgumentError: If GraphQL syntax validation fails on the server.
+            PermissionDeniedError: If an @auth policy directive blocks execution due to
+                insufficient permission.
+            NotFoundError: If a specified resource is not found, or the request is rejected
+                by undisclosed reasons, such as whitelisting.
+            InternalError: If the server response payload is invalid or malformed.
+            FirebaseError: The base platform exception.
+        """
+        raise NotImplementedError
+
+    def execute_graphql_read(
+        self,
+        query: str,
+        options: Optional[GraphqlOptions[_Variables]] = None,
+        variables_type: Type[_Variables] = Any,
+    ) -> ExecuteGraphqlResponse[Any]:
+        """Executes a read-only GraphQL query and returns the result.
+
+        Args:
+            query: string containing the read-only GraphQL query
+            options: GraphqlOptions instance containing operational parameters such as
+                variables, operation name, or impersonation context (optional).
+            variables_type: The expected structure for the request variables
+
+        Returns:
+            ExecuteGraphqlResponse: An ExecuteGraphqlResponse containing the raw
+                response data dictionary.
+
+        Raises:
+            ValueError: If the arguments are invalid from the local inputs side.
+            InvalidArgumentError: If GraphQL syntax validation fails on the server.
+            PermissionDeniedError: If an @auth policy directive blocks execution due to
+                insufficient permission.
+            NotFoundError: If a specified resource is not found, or the request is rejected
+                by undisclosed reasons, such as whitelisting.
+            InternalError: If the server response payload is invalid or malformed.
+            FirebaseError: The base platform exception.
+        """
+        raise NotImplementedError
+
 
 class _DataConnectService:
     """Service that maintains a collection of DataConnect clients."""
@@ -170,42 +302,6 @@ def client(config: ConnectorConfig, app: Optional[App] = None) -> DataConnect:
     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
-
-
-# TODO(b/406281627): Add support for partial errors.
-@dataclass
-class ExecuteGraphqlResponse(Generic[_Data]):
-    """Represents the response from a DataConnect GraphQL execution.
-
-    Attributes:
-        data: The raw JSON dictionary returned by the GraphQL execution.
-    """
-    data: _Data
-
-
-
 def _get_emulator_host() -> Optional[str]:
     return _utils.get_emulator_host("DATA_CONNECT_EMULATOR_HOST")
 
@@ -421,3 +517,32 @@ def _parse_graphql_response(
 
         # TODO(b/406281627): Add support for partial errors.
         return ExecuteGraphqlResponse(data=resp_dict.get("data"))
+
+
+    def _execute_graphql_helper(
+        self,
+        query: str,
+        endpoint: str,
+        options: Optional[GraphqlOptions[_Variables]] = None,
+        variables_type: Type[_Variables] = Any,
+    ) -> ExecuteGraphqlResponse[Any]:
+        """Helper method to execute GraphQL queries or mutations against a specified endpoint."""
+        raise NotImplementedError
+
+    def execute_graphql(
+        self,
+        query: str,
+        options: Optional[GraphqlOptions[_Variables]] = None,
+        variables_type: Type[_Variables] = Any,
+    ) -> ExecuteGraphqlResponse[Any]:
+        """Executes a GraphQL query or mutation and returns the result."""
+        raise NotImplementedError
+
+    def execute_graphql_read(
+        self,
+        query: str,
+        options: Optional[GraphqlOptions[_Variables]] = None,
+        variables_type: Type[_Variables] = Any,
+    ) -> ExecuteGraphqlResponse[Any]:
+        """Executes a read-only GraphQL query and returns the result."""
+        raise NotImplementedError