feat(fdc): Enforce Mapping or dataclass for variables and type-annotate variable_type

- Updated _validate_variables_type to validate that if variables are provided, they must be either a collections.abc.Mapping or a dataclass.

- Type-annotated the variable_type parameter as Optional[Type[Any]] = None in both validation helper signatures.

- Refactored variable unit tests to use a realistic CreateUserVariables (nesting a UserProfile dataclass) instead of response-like shapes.

- Added a test case confirming that standard dictionaries (Mapping) are accepted as valid variables.

Author: mk2023

firebase_admin/dataconnect.py

@@ -18,8 +18,9 @@
 Firebase apps.
 """
 
+from collections.abc import Mapping
 from dataclasses import dataclass, asdict, is_dataclass
-from typing import Any, Dict, Generic, Optional, TypeVar, Union
+from typing import Any, Dict, Generic, Optional, Type, TypeVar, Union
 import firebase_admin
 from firebase_admin import _utils, _http_client, App
 
@@ -214,11 +215,18 @@ def __init__(self, connector_config: ConnectorConfig, app: App) -> None:
 
         self._http_client = _http_client.JsonHttpClient(credential=self._credential)
 
-    def _validate_variables_type(self, variables: Any, variable_type: Any) -> None:
+    def _validate_variables_type(
+        self,
+        variables: Any,
+        variable_type: Optional[Type[Any]] = None
+    ) -> None:
         """Validates variables against expected type."""
-        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__}")
+        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:
+                if not isinstance(variables, variable_type):
+                    raise ValueError(f"variables must be of type {variable_type.__name__}")
 
     def _validate_impersonation_options(self, impersonate: Any) -> None:
         """Validates impersonation dictionary options."""
@@ -245,7 +253,7 @@ def _validate_impersonation_options(self, impersonate: Any) -> None:
     def _validate_graphql_options(
         self,
         graphql_options: Optional[GraphqlOptions[Any]],
-        variable_type: Any = None
+        variable_type: Optional[Type[Any]] = None
     ) -> None:
         """Validates GraphqlOptions inputs at runtime."""
         if graphql_options is not None:

tests/test_data_connect.py

@@ -399,8 +399,8 @@ def test_validate_graphql_options_valid(self):
         # Valid with no options
         self.api_client._validate_graphql_options(None)
 
-        # Valid with options
-        options = dataconnect.GraphqlOptions(variables={"foo": "bar"})
+        # Valid with default options (no arguments)
+        options = dataconnect.GraphqlOptions()
         self.api_client._validate_graphql_options(options)
 
     def test_validate_graphql_options_valid_impersonate(self):
@@ -416,21 +416,28 @@ def test_validate_graphql_options_valid_impersonate(self):
         options = dataconnect.GraphqlOptions(impersonate=imp_auth)
         self.api_client._validate_graphql_options(options)
 
-    def test_validate_graphql_options_valid_variables(self):
+    def test_validate_graphql_options_valid_dataclass_variables(self):
         @dataclass
-        class User:
-            user_id: str
-            name: str
+        class UserProfile:
             address: str
+            phone: str
 
         @dataclass
-        class UsersResponse:
-            users: list[User]
+        class CreateUserVariables:
+            user_id: str
+            name: str
+            profile: UserProfile
 
-        users_val = [User(user_id="1", name="Fred", address="123 Road")]
-        valid_variables = UsersResponse(users=users_val)
+        profile_val = UserProfile(address="123 Road", phone="332-3233-0199")
+        valid_variables = CreateUserVariables(
+            user_id="1", name="Fred", profile=profile_val
+        )
         options = dataconnect.GraphqlOptions(variables=valid_variables)
-        self.api_client._validate_graphql_options(options, UsersResponse)
+        self.api_client._validate_graphql_options(options, CreateUserVariables)
+
+    def test_validate_graphql_options_valid_mapping_variables(self):
+        options = dataconnect.GraphqlOptions(variables={"user_id": "1", "name": "Fred"})
+        self.api_client._validate_graphql_options(options)
 
     def test_validate_graphql_options_invalid_options(self):
         with pytest.raises(ValueError, match="options must be a GraphqlOptions instance"):
@@ -490,18 +497,26 @@ def test_validate_graphql_options_invalid_operation_name(self):
 
     def test_validate_graphql_options_invalid_variables(self):
         @dataclass
-        class User:
-            user_id: str
-            name: str
+        class UserProfile:
             address: str
+            phone: str
 
         @dataclass
-        class UsersResponse:
-            users: list[User]
+        class CreateUserVariables:
+            user_id: str
+            name: str
+            profile: UserProfile
+
+        # Test invalid variable format (not Mapping or dataclass)
+        options = dataconnect.GraphqlOptions(variables="invalid-string-format")
+        msg = "variables must be a collections.abc.Mapping or a dataclass"
+        with pytest.raises(ValueError, match=msg):
+            self.api_client._validate_graphql_options(options)
 
-        options = dataconnect.GraphqlOptions(variables="not-users-response")
-        with pytest.raises(ValueError, match="variables must be of type UsersResponse"):
-            self.api_client._validate_graphql_options(options, UsersResponse)
+        # Test valid Mapping format but type mismatch against expected dataclass type
+        options = dataconnect.GraphqlOptions(variables={"foo": "bar"})
+        with pytest.raises(ValueError, match="variables must be of type CreateUserVariables"):
+            self.api_client._validate_graphql_options(options, CreateUserVariables)
 
 
 class TestDataConnectApiClientPrepareGraphqlPayload:
@@ -529,29 +544,31 @@ def test_prepare_graphql_payload_with_variables(self):
 
     def test_prepare_graphql_payload_with_dataclass_variables(self):
         @dataclass
-        class User:
-            user_id: str
-            name: str
+        class UserProfile:
             address: str
+            phone: str
 
         @dataclass
-        class UsersResponse:
-            users: list[User]
+        class CreateUserVariables:
+            user_id: str
+            name: str
+            profile: UserProfile
 
-        users_val = [User(user_id="1", name="Fred", address="123 Road")]
-        valid_variables = UsersResponse(users=users_val)
+        profile_val = UserProfile(address="123 Road", phone="332-3233-0199")
+        valid_variables = CreateUserVariables(
+            user_id="1", name="Fred", profile=profile_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"
-                    }
-                ]
+                "user_id": "1",
+                "name": "Fred",
+                "profile": {
+                    "address": "123 Road",
+                    "phone": "332-3233-0199"
+                }
             }
         }
 
@@ -589,17 +606,20 @@ def test_prepare_graphql_payload_with_impersonate_authenticated(self):
 
     def test_prepare_graphql_payload_with_all_fields(self):
         @dataclass
-        class User:
-            user_id: str
-            name: str
+        class UserProfile:
             address: str
+            phone: str
 
         @dataclass
-        class UsersResponse:
-            users: list[User]
+        class CreateUserVariables:
+            user_id: str
+            name: str
+            profile: UserProfile
 
-        users_val = [User(user_id="1", name="Fred", address="123 Road")]
-        valid_variables = UsersResponse(users=users_val)
+        profile_val = UserProfile(address="123 Road", phone="332-3233-0199")
+        valid_variables = CreateUserVariables(
+            user_id="1", name="Fred", profile=profile_val
+        )
         imp_auth = dataconnect.Impersonation.authenticated(
             {"sub": "authenticated-UUID"}
         )
@@ -613,13 +633,12 @@ class UsersResponse:
             "query": "query { hello }",
             "operationName": "getUsers",
             "variables": {
-                "users": [
-                    {
-                        "user_id": "1",
-                        "name": "Fred",
-                        "address": "123 Road"
-                    }
-                ]
+                "user_id": "1",
+                "name": "Fred",
+                "profile": {
+                    "address": "123 Road",
+                    "phone": "332-3233-0199"
+                }
             },
             "extensions": {
                 "impersonate": {"authClaims": {"sub": "authenticated-UUID"}}