refactor: extract _content_fields ClassVar to eliminate duplicate _build_metadata() overrides

.github/workflows/claude-code-review.yml

@@ -38,7 +38,7 @@ jobs:
           claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }}
           plugin_marketplaces: 'https://github.com/anthropics/claude-code.git'
           plugins: 'code-review@claude-code-plugins'
-          claude_args: '--model claude-opus-4-6'
+          claude_args: '--model claude-sonnet-4-6'
           prompt: '/code-review:code-review --comment ${{ github.repository }}/pull/${{ github.event.pull_request.number }}'
           # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md
           # or https://code.claude.com/docs/en/cli-reference for available options

src/celeste/client.py

@@ -44,6 +44,7 @@ class OpenAITextClient(OpenAIResponsesMixin, TextClient):
     model: Model
     auth: Authentication
     provider: Provider
+    _content_fields: ClassVar[set[str]] = set()
 
     @property
     @abstractmethod
@@ -302,6 +303,10 @@ def _validate_artifacts(
 
     def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
         """Build metadata dictionary from response data."""
+        if self._content_fields:
+            response_data = {
+                k: v for k, v in response_data.items() if k not in self._content_fields
+            }
         return {
             "model": self.model.id,
             "provider": self.provider,

src/celeste/protocols/chatcompletions/client.py

@@ -23,7 +23,7 @@ class ChatCompletionsClient(APIMixin):
     - _parse_usage() - Extract usage dict from response
     - _parse_content() - Extract choices array from response
     - _parse_finish_reason() - Extract finish reason from response
-    - _build_metadata() - Filter content fields
+    - _content_fields: ClassVar - Content field names to exclude from metadata
 
     Providers override ClassVars and hook methods:
     - _default_base_url: ClassVar[str] - Provider's API base URL
@@ -39,6 +39,7 @@ class DeepSeekChatClient(ChatCompletionsClient):
 
     _default_base_url: ClassVar[str] = config.DEFAULT_BASE_URL
     _default_endpoint: ClassVar[str] = config.ChatCompletionsEndpoint.CREATE_CHAT
+    _content_fields: ClassVar[set[str]] = {"choices"}
 
     def _build_url(self, endpoint: str, streaming: bool = False) -> str:
         """Build full URL for request.
@@ -150,13 +151,5 @@ def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason:
             reason = choices[0].get("finish_reason")
         return FinishReason(reason=reason)
 
-    def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
-        """Build metadata dictionary, filtering out content fields."""
-        content_fields = {"choices"}
-        filtered_data = {
-            k: v for k, v in response_data.items() if k not in content_fields
-        }
-        return super()._build_metadata(filtered_data)
-
 
 __all__ = ["ChatCompletionsClient"]

src/celeste/protocols/openresponses/client.py

@@ -23,7 +23,7 @@ class OpenResponsesClient(APIMixin):
     - _parse_usage() - Extract usage dict from response
     - _parse_content() - Extract output array from response
     - _parse_finish_reason() - Extract finish reason from response
-    - _build_metadata() - Filter content fields
+    - _content_fields: ClassVar - Content field names to exclude from metadata
 
     Providers override ClassVars and hook methods:
     - _default_base_url: ClassVar[str] - Provider's API base URL
@@ -39,6 +39,7 @@ class OpenAIResponsesClient(OpenResponsesClient):
 
     _default_base_url: ClassVar[str] = config.DEFAULT_BASE_URL
     _default_endpoint: ClassVar[str] = config.OpenResponsesEndpoint.CREATE_RESPONSE
+    _content_fields: ClassVar[set[str]] = {"output"}
 
     def _build_url(self, endpoint: str, streaming: bool = False) -> str:
         """Build full URL for request.
@@ -152,13 +153,5 @@ def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason:
                     return FinishReason(reason="completed")
         return FinishReason(reason=None)
 
-    def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
-        """Build metadata dictionary, filtering out content fields."""
-        content_fields = {"output"}
-        filtered_data = {
-            k: v for k, v in response_data.items() if k not in content_fields
-        }
-        return super()._build_metadata(filtered_data)
-
 
 __all__ = ["OpenResponsesClient"]

src/celeste/providers/anthropic/messages/client.py

@@ -1,7 +1,7 @@
 """Anthropic Messages API client mixin."""
 
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, ClassVar
 
 from celeste.client import APIMixin
 from celeste.core import UsageField
@@ -21,7 +21,7 @@ class AnthropicMessagesClient(APIMixin):
     - _parse_usage() - Extract usage dict from response
     - _parse_content() - Extract content array from response
     - _parse_finish_reason() - Extract finish reason from response
-    - _build_metadata() - Filter content fields
+    - _content_fields: ClassVar - Content field names to exclude from metadata
 
     Auth-based endpoint selection:
     - GoogleADC auth -> Vertex AI endpoints (Claude on Google Cloud)
@@ -37,6 +37,8 @@ def _parse_content(self, response_data, **parameters):
                 return ""
     """
 
+    _content_fields: ClassVar[set[str]] = {"content"}
+
     def _get_vertex_endpoint(
         self, anthropic_endpoint: str, streaming: bool = False
     ) -> str:
@@ -185,13 +187,5 @@ def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason:
         stop_reason = response_data.get("stop_reason")
         return FinishReason(reason=stop_reason)
 
-    def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
-        """Build metadata dictionary, filtering out content fields."""
-        content_fields = {"content"}
-        filtered_data = {
-            k: v for k, v in response_data.items() if k not in content_fields
-        }
-        return super()._build_metadata(filtered_data)
-
 
 __all__ = ["AnthropicMessagesClient"]

src/celeste/providers/bfl/images/client.py

@@ -3,7 +3,7 @@
 import asyncio
 import time
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, ClassVar
 
 from celeste.client import APIMixin
 from celeste.core import UsageField
@@ -33,6 +33,8 @@ def _parse_content(self, response_data, **parameters):
                 # Extract image from result["sample"]...
     """
 
+    _content_fields: ClassVar[set[str]] = {"result"}
+
     async def _make_request(
         self,
         request_body: dict[str, Any],
@@ -152,13 +154,5 @@ def _parse_content(self, response_data: dict[str, Any]) -> Any:
             raise ValueError(msg)
         return result
 
-    def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
-        """Build metadata dictionary, filtering out content fields."""
-        content_fields = {"result"}
-        filtered_data = {
-            k: v for k, v in response_data.items() if k not in content_fields
-        }
-        return super()._build_metadata(filtered_data)
-
 
 __all__ = ["BFLImagesClient"]

src/celeste/providers/byteplus/images/client.py

@@ -1,7 +1,7 @@
 """BytePlus Images API client mixin."""
 
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, ClassVar
 
 from celeste.client import APIMixin
 from celeste.core import UsageField
@@ -20,7 +20,7 @@ class BytePlusImagesClient(APIMixin):
     - _parse_usage() - Extract usage dict from response
     - _parse_content() - Extract images/data array from response
     - _parse_finish_reason() - Extract finish reason from response
-    - _build_metadata() - Filter content fields and extract seed
+    - _content_fields: ClassVar - Content field names to exclude from metadata
 
     Usage:
         class BytePlusImageGenerationClient(BytePlusImagesClient, ImageGenerationClient):
@@ -29,6 +29,8 @@ def _parse_content(self, response_data, **parameters):
                 # Extract image from images[0] or data[0]...
     """
 
+    _content_fields: ClassVar[set[str]] = {"images", "data"}
+
     def _build_request(
         self,
         inputs: Any,
@@ -135,18 +137,10 @@ def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason:
 
     def _build_metadata(self, response_data: dict[str, Any]) -> Any:
         """Build metadata dictionary, extracting seed if present."""
-        # Filter content fields before calling super
-        content_fields = {"images", "data"}
-        filtered_data = {
-            k: v for k, v in response_data.items() if k not in content_fields
-        }
-        metadata = super()._build_metadata(filtered_data)
-
-        # Add provider-specific parsed fields
+        metadata = super()._build_metadata(response_data)
         seed = response_data.get("seed")
         if seed is not None:
             metadata["seed"] = seed
-
         return metadata
 
 

src/celeste/providers/byteplus/videos/client.py

@@ -4,7 +4,7 @@
 import logging
 import time
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, ClassVar
 
 from celeste.client import APIMixin
 from celeste.core import UsageField
@@ -35,6 +35,8 @@ def _parse_content(self, response_data, **parameters):
                 # Extract video from content["video_url"]...
     """
 
+    _content_fields: ClassVar[set[str]] = {"content"}
+
     def _build_request(
         self,
         inputs: Any,
@@ -167,13 +169,5 @@ def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason:
         """BytePlus provides status but not structured finish reasons."""
         return FinishReason(reason=None)
 
-    def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
-        """Build metadata dictionary, filtering out content fields."""
-        content_fields = {"content"}
-        filtered_data = {
-            k: v for k, v in response_data.items() if k not in content_fields
-        }
-        return super()._build_metadata(filtered_data)
-
 
 __all__ = ["BytePlusVideosClient"]

src/celeste/providers/cohere/chat/client.py

@@ -1,7 +1,7 @@
 """Cohere Chat API client mixin."""
 
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, ClassVar
 
 from celeste.client import APIMixin
 from celeste.core import UsageField
@@ -20,7 +20,7 @@ class CohereChatClient(APIMixin):
     - _parse_usage() - Extract usage dict from response
     - _parse_content() - Extract content array from response message
     - _parse_finish_reason() - Extract finish reason from response
-    - _build_metadata() - Filter content fields
+    - _content_fields: ClassVar - Content field names to exclude from metadata
 
     Usage:
         class CohereTextGenerationClient(CohereChatClient, TextGenerationClient):
@@ -30,6 +30,8 @@ def _parse_content(self, response_data, **parameters):
                 return self._transform_output(text, **parameters)
     """
 
+    _content_fields: ClassVar[set[str]] = {"message"}
+
     def _build_request(
         self,
         inputs: Any,
@@ -132,13 +134,5 @@ def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason:
         reason = response_data.get("finish_reason")
         return FinishReason(reason=reason)
 
-    def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
-        """Build metadata dictionary, filtering out content fields."""
-        content_fields = {"message"}
-        filtered_data = {
-            k: v for k, v in response_data.items() if k not in content_fields
-        }
-        return super()._build_metadata(filtered_data)
-
 
 __all__ = ["CohereChatClient"]

src/celeste/providers/elevenlabs/text_to_speech/client.py

@@ -165,10 +165,6 @@ def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason:
         """ElevenLabs TTS doesn't provide finish reasons."""
         return FinishReason(reason=None)
 
-    def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
-        """Build metadata dictionary, filtering out content fields."""
-        return super()._build_metadata(response_data)
-
     def _map_output_format_to_mime_type(
         self, output_format: str | None
     ) -> AudioMimeType:

src/celeste/providers/google/cloud_tts/client.py

@@ -1,7 +1,7 @@
 """Google Cloud TTS API client mixin."""
 
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, ClassVar
 
 from celeste.client import APIMixin
 from celeste.exceptions import StreamingNotSupportedError
@@ -30,6 +30,8 @@ def _parse_content(self, response_data, **params):
                 return AudioArtifact(data=audio_bytes, mime_type=..., ...)
     """
 
+    _content_fields: ClassVar[set[str]] = {"audioContent"}
+
     def _make_stream_request(
         self,
         request_body: dict[str, Any],
@@ -100,13 +102,5 @@ def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason:
         """Cloud TTS API doesn't provide finish reasons."""
         return FinishReason(reason=None)
 
-    def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
-        """Build metadata dictionary, filtering out content fields."""
-        content_fields = {"audioContent"}
-        filtered_data = {
-            k: v for k, v in response_data.items() if k not in content_fields
-        }
-        return super()._build_metadata(filtered_data)
-
 
 __all__ = ["GoogleCloudTTSClient"]

src/celeste/providers/google/embeddings/client.py

@@ -1,7 +1,7 @@
 """Google Embeddings API client mixin."""
 
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, ClassVar
 
 from celeste.client import APIMixin
 from celeste.exceptions import StreamingNotSupportedError
@@ -29,6 +29,8 @@ def _parse_content(self, response_data, **params):
                 return super()._parse_content(response_data)  # No transformation needed
     """
 
+    _content_fields: ClassVar[set[str]] = {"embedding", "embeddings"}
+
     def _make_stream_request(
         self,
         request_body: dict[str, Any],
@@ -141,13 +143,5 @@ def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason:
         """Embeddings API doesn't provide finish reasons."""
         return FinishReason(reason=None)
 
-    def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
-        """Build metadata, filtering out embedding content."""
-        content_fields = {"embedding", "embeddings"}
-        filtered_data = {
-            k: v for k, v in response_data.items() if k not in content_fields
-        }
-        return super()._build_metadata(filtered_data)
-
 
 __all__ = ["GoogleEmbeddingsClient"]

src/celeste/providers/google/generate_content/client.py

@@ -1,7 +1,7 @@
 """Google GenerateContent API client mixin."""
 
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, ClassVar
 
 from celeste.client import APIMixin
 from celeste.core import UsageField
@@ -21,7 +21,7 @@ class GoogleGenerateContentClient(APIMixin):
     - _parse_usage() - Extract usage dict from usageMetadata
     - _parse_content() - Extract parts array from first candidate
     - _parse_finish_reason() - Extract finish reason string from candidates
-    - _build_metadata() - Filter content fields
+    - _content_fields: ClassVar - Content field names to exclude from metadata
 
     Auth-based endpoint selection:
     - GoogleADC auth -> Vertex AI endpoints
@@ -37,6 +37,8 @@ def _parse_content(self, response_data, **parameters):
                 return self._transform_output(text, **parameters)
     """
 
+    _content_fields: ClassVar[set[str]] = {"candidates"}
+
     def _get_vertex_endpoint(self, gemini_endpoint: str) -> str:
         """Map Gemini endpoint to Vertex AI endpoint."""
         mapping: dict[str, str] = {
@@ -145,13 +147,5 @@ def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason:
             reason = candidates[0].get("finishReason")
         return FinishReason(reason=reason)
 
-    def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]:
-        """Build metadata dictionary, filtering out content fields."""
-        content_fields = {"candidates"}
-        filtered_data = {
-            k: v for k, v in response_data.items() if k not in content_fields
-        }
-        return super()._build_metadata(filtered_data)
-
 
 __all__ = ["GoogleGenerateContentClient"]