ci: Add google-genai to uv typing group (#7323)

Add `google-genai` to the typing dependency group and fix the resulting mypy errors.

Author: alexander-alderman-webb

pyproject.toml

@@ -74,6 +74,7 @@ typing = [
     "pydantic-ai-slim>=2.23.0",
     "langchain-core>=1.5.3",
     "huggingface-hub>=1.26.1",
+    "google-genai>=2.21.0",
 ]
 test = [
     "dataclasses ; python_full_version < '3.7'",

sentry_sdk/integrations/google_genai/__init__.py

@@ -47,20 +47,20 @@ def __init__(self: "GoogleGenAIIntegration", include_prompts: bool = True) -> No
     @staticmethod
     def setup_once() -> None:
         # Patch sync methods
-        Models.generate_content = _wrap_generate_content(Models.generate_content)
-        Models.generate_content_stream = _wrap_generate_content_stream(
+        Models.generate_content = _wrap_generate_content(Models.generate_content)  # type: ignore[method-assign]
+        Models.generate_content_stream = _wrap_generate_content_stream(  # type: ignore[method-assign]
             Models.generate_content_stream
         )
-        Models.embed_content = _wrap_embed_content(Models.embed_content)
+        Models.embed_content = _wrap_embed_content(Models.embed_content)  # type: ignore[method-assign]
 
         # Patch async methods
-        AsyncModels.generate_content = _wrap_async_generate_content(
+        AsyncModels.generate_content = _wrap_async_generate_content(  # type: ignore[method-assign]
             AsyncModels.generate_content
         )
-        AsyncModels.generate_content_stream = _wrap_async_generate_content_stream(
+        AsyncModels.generate_content_stream = _wrap_async_generate_content_stream(  # type: ignore[method-assign]
             AsyncModels.generate_content_stream
         )
-        AsyncModels.embed_content = _wrap_async_embed_content(AsyncModels.embed_content)
+        AsyncModels.embed_content = _wrap_async_embed_content(AsyncModels.embed_content)  # type: ignore[method-assign]
 
 
 def _wrap_generate_content_stream(f: "Callable[..., Any]") -> "Callable[..., Any]":

sentry_sdk/integrations/google_genai/utils.py

@@ -13,6 +13,7 @@
     Optional,
     TypedDict,
     Union,
+    cast,
 )
 
 from google.genai.types import Content, GenerateContentConfig, Part, PartDict
@@ -44,13 +45,15 @@
 
 if TYPE_CHECKING:
     from google.genai.types import (
+        ContentDict,
         ContentListUnion,
         ContentUnion,
         ContentUnionDict,
         EmbedContentResponse,
         GenerateContentResponse,
         Model,
         Tool,
+        ToolUnion,
     )
 
     from sentry_sdk._types import TextPart
@@ -555,7 +558,7 @@ def extract_contents_text(contents: "ContentListUnion") -> "Optional[str]":
 
 
 def _format_tools_for_span(
-    tools: "Iterable[Tool | Callable[..., Any]]",
+    tools: "Iterable[ToolUnion]",
 ) -> "Optional[List[dict[str, Any]]]":
     """Format tools parameter for span data."""
     formatted_tools = []
@@ -604,13 +607,17 @@ def extract_tool_calls(
     tool_calls = []
 
     # Extract from candidates, sometimes tool calls are nested under the content.parts object
-    if getattr(response, "candidates", []):
-        for candidate in response.candidates:
+    candidates = getattr(response, "candidates", [])
+    if response is not None:
+        for candidate in candidates:
             if not hasattr(candidate, "content") or not getattr(
                 candidate.content, "parts", []
             ):
                 continue
 
+            if candidate.content is None or candidate.content.parts is None:
+                continue
+
             for part in candidate.content.parts:
                 if getattr(part, "function_call", None):
                     function_call = part.function_call
@@ -627,34 +634,42 @@ def extract_tool_calls(
 
     # Extract from automatic_function_calling_history
     # This is the history of tool calls made by the model
-    if getattr(response, "automatic_function_calling_history", None):
-        for content in response.automatic_function_calling_history:
-            if not getattr(content, "parts", None):
-                continue
+    automatic_function_calling_history = getattr(
+        response, "automatic_function_calling_history", None
+    )
+    if automatic_function_calling_history is None:
+        return tool_calls if tool_calls else None
 
-            for part in getattr(content, "parts", []):
-                if getattr(part, "function_call", None):
-                    function_call = part.function_call
-                    tool_call = {
-                        "name": getattr(function_call, "name", None),
-                        "type": "function_call",
-                    }
+    for content in automatic_function_calling_history:
+        if not getattr(content, "parts", None):
+            continue
 
-                    # Extract arguments if available
-                    if hasattr(function_call, "args"):
-                        tool_call["arguments"] = safe_serialize(function_call.args)
+        for part in getattr(content, "parts", []):
+            if getattr(part, "function_call", None):
+                function_call = part.function_call
+                tool_call = {
+                    "name": getattr(function_call, "name", None),
+                    "type": "function_call",
+                }
 
-                    tool_calls.append(tool_call)
+                # Extract arguments if available
+                if hasattr(function_call, "args"):
+                    tool_call["arguments"] = safe_serialize(function_call.args)
+
+                tool_calls.append(tool_call)
 
     return tool_calls if tool_calls else None
 
 
 def _capture_tool_input(
-    args: "tuple[Any, ...]", kwargs: "dict[str, Any]", tool: "Tool"
+    args: "tuple[Any, ...]", kwargs: "dict[str, Any]", tool: "Tool | Callable[..., Any]"
 ) -> "dict[str, Any]":
     """Capture tool input from args and kwargs."""
     tool_input = kwargs.copy() if kwargs else {}
 
+    if not callable(tool):
+        return tool_input
+
     # If we have positional args, try to map them to the function signature
     if args:
         try:
@@ -767,29 +782,35 @@ def sync_wrapped(*args: "Any", **kwargs: "Any") -> "Any":
 
 
 def wrapped_config_with_tools(
-    config: "GenerateContentConfig",
-) -> "GenerateContentConfig":
+    config: "GenerateContentConfig | None",
+) -> "GenerateContentConfig | None":
     """Wrap tools in config to emit execute_tool spans. Tools are sometimes passed directly as
     callable functions as a part of the config object."""
+    if not config:
+        return config
 
-    if not config or not getattr(config, "tools", None):
+    tools = getattr(config, "tools", None)
+    if tools is None:
         return config
 
     result = copy.copy(config)
-    result.tools = [wrapped_tool(tool) for tool in config.tools]
+    result.tools = [wrapped_tool(tool) for tool in tools]
 
     return result
 
 
 def _extract_response_text(
     response: "GenerateContentResponse",
-) -> "Optional[List[str]]":
+) -> "Optional[List[str | None]]":
     """Extract text from response candidates."""
 
     if not response or not getattr(response, "candidates", []):
         return None
 
-    texts = []
+    texts: "list[str | None]" = []
+    if response.candidates is None:
+        return texts if texts else None
+
     for candidate in response.candidates:
         if not hasattr(candidate, "content") or not hasattr(candidate.content, "parts"):
             continue
@@ -811,7 +832,11 @@ def extract_finish_reasons(
     if not response or not getattr(response, "candidates", []):
         return None
 
-    finish_reasons = []
+    finish_reasons: "list[str]" = []
+
+    if response.candidates is None:
+        return finish_reasons if finish_reasons else None
+
     for candidate in response.candidates:
         if getattr(candidate, "finish_reason", None):
             # Convert enum value to string if necessary
@@ -843,16 +868,29 @@ def _transform_system_instruction_one_level(
                     text_parts.append({"type": "text", "content": part.text})
         return text_parts
 
-    if isinstance(system_instructions, dict) and system_instructions.get("text"):
-        return [{"type": "text", "content": system_instructions["text"]}]
+    if isinstance(system_instructions, dict) and "text" in system_instructions:
+        text = cast("PartDict", system_instructions)["text"]
+        if text is None:
+            return []
+
+        return [{"type": "text", "content": text}]
 
     elif can_be_content and isinstance(system_instructions, dict):
-        parts = system_instructions.get("parts", [])
+        parts = cast("ContentDict", system_instructions).get("parts", [])
+        if parts is None:
+            return text_parts
+
         for part in parts:
             if isinstance(part, Part) and isinstance(part.text, str):
                 text_parts.append({"type": "text", "content": part.text})
-            elif isinstance(part, dict) and isinstance(part.get("text"), str):
-                text_parts.append({"type": "text", "content": part["text"]})
+                continue
+
+            if not isinstance(part, dict):
+                continue
+
+            text = part.get("text")
+            if isinstance(text, str):
+                text_parts.append({"type": "text", "content": text})
         return text_parts
 
     return text_parts
@@ -1002,11 +1040,13 @@ def set_span_data_for_response(
             span, SPANDATA.GEN_AI_RESPONSE_FINISH_REASONS, finish_reasons
         )
 
-    if getattr(response, "response_id", None):
-        set_on_span(SPANDATA.GEN_AI_RESPONSE_ID, response.response_id)
+    response_id = getattr(response, "response_id", None)
+    if response_id is not None:
+        set_on_span(SPANDATA.GEN_AI_RESPONSE_ID, response_id)
 
-    if getattr(response, "model_version", None):
-        set_on_span(SPANDATA.GEN_AI_RESPONSE_MODEL, response.model_version)
+    model_version = getattr(response, "model_version", None)
+    if model_version is not None:
+        set_on_span(SPANDATA.GEN_AI_RESPONSE_MODEL, model_version)
 
     usage_data = extract_usage_data(response)
 
@@ -1065,7 +1105,7 @@ def prepare_generate_content_args(
     contents = args[1] if len(args) > 1 else kwargs.get("contents")
     model_name = get_model_name(model)
 
-    config = kwargs.get("config")
+    config: "GenerateContentConfig | None" = kwargs.get("config")
     wrapped_config = wrapped_config_with_tools(config)
     if wrapped_config is not config:
         kwargs["config"] = wrapped_config