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

[mk2023]

Added execute_graphql and execute_graphql_read method signatures and docstrings to DataConnect and _DataConnectApiClient. Expanded unit test coverage in tests/test_data_connect.py for payload formatting, options validation, and error bubbling, and introduced a comprehensive integration test suite in integration/test_data_connect.py translated from Node.js Admin SDK integration tests.

[gemini-code-assist]

Code Review

This pull request introduces execute_graphql and execute_graphql_read methods to the DataConnect client, along with corresponding integration and unit tests. The review feedback highlights that several of these new methods are left unimplemented (raising NotImplementedError) and provides their implementation details. Additionally, the reviewer identifies a critical type-checking bug in the validation logic when variables_type is Any, and points out a mismatch in a test assertion message.

[gemini-code-assist]

The execute_graphql method currently raises NotImplementedError. It should be implemented to delegate the execution to the internal API client self._client.

    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.
        """
        return self._client.execute_graphql(
            query=query,
            options=options,
            variables_type=variables_type,
        )

[gemini-code-assist]

The execute_graphql_read method currently raises NotImplementedError. It should be implemented to delegate the execution to the internal API client self._client.

    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.
        """
        return self._client.execute_graphql_read(
            query=query,
            options=options,
            variables_type=variables_type,
        )

[gemini-code-assist]

These methods in _DataConnectApiClient currently raise NotImplementedError. They should be implemented using the existing helper methods to construct the request, send it, and parse the response.

Additionally, there is a critical bug in the existing _validate_variables_type method (which is called by _validate_graphql_options). When variables_type is left as its default value Any, and options contains variables, _validate_variables_type will perform isinstance(variables, expected_type) where expected_type is typing.Any. In Python, isinstance(obj, typing.Any) raises a TypeError: Instance and class checks can only be used with types with unique class checks.

To fix this, please also update _validate_variables_type (around line 309) to skip the isinstance check if variable_type is typing.Any:

if variable_type is not None and variable_type is not typing.Any:
    expected_type = typing.get_origin(variable_type) or variable_type
    if not isinstance(variables, expected_type):
        ...

    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."""
        if not isinstance(query, str):
            raise ValueError("query must be a string")
        if not query.strip():
            raise ValueError("query must be a non-empty string")
        self._validate_graphql_options(options, variables_type)

        url = self._get_firebase_dataconnect_service_url(endpoint)
        headers = self._get_headers()
        payload = self._prepare_graphql_payload(query, options)

        resp_dict = self._make_gql_request(url, headers, payload)
        return self._parse_graphql_response(resp_dict)

    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."""
        return self._execute_graphql_helper(
            query=query,
            endpoint=_EXECUTE_GRAPHQL_ENDPOINT,
            options=options,
            variables_type=variables_type,
        )

    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."""
        return self._execute_graphql_helper(
            query=query,
            endpoint=_EXECUTE_GRAPHQL_READ_ENDPOINT,
            options=options,
            variables_type=variables_type,
        )