Removed Magic Classes logic implemented directly on BinaryContent and DocumentUrl

pulphix · pulphix · commit 83b8e15809fe · 2025-09-15T09:04:56.000+02:00
diff --git a/examples/pydantic_ai_examples/textlike_file_mapping.py b/examples/pydantic_ai_examples/textlike_file_mapping.py
@@ -9,7 +9,8 @@
 
 from __future__ import annotations
 
-from pydantic_ai import Agent, MagicBinaryContent, MagicDocumentUrl
+from pydantic_ai import Agent
+from pydantic_ai.messages import BinaryContent, DocumentUrl
 
 # Load API keys from .env if available
 try:  # pragma: no cover - example bootstrap
@@ -24,24 +25,21 @@ def run_with_openai() -> None:
     agent = Agent('openai:gpt-4o')
 
     # Text file by URL → becomes inline text with a file delimiter on OpenAI
-    txt_url = MagicDocumentUrl(
+    txt_url = DocumentUrl(
         url='https://raw.githubusercontent.com/pydantic/pydantic/main/README.md',
-        filename='README.md',
         # media_type optional; inferred from extension if omitted
         media_type='text/plain',
     )
 
     # Binary text (bytes) → becomes inline text with a file delimiter on OpenAI
-    txt_bytes = MagicBinaryContent(
+    txt_bytes = BinaryContent(
         data=b'Hello from bytes',
         media_type='text/plain',
-        filename='hello.txt',
     )
 
     # PDF by URL → remains a file part (base64 + strict MIME) on OpenAI
-    pdf_url = MagicDocumentUrl(
+    pdf_url = DocumentUrl(
         url='https://arxiv.org/pdf/2403.05530.pdf',
-        filename='gemini-tech-report.pdf',
         media_type='application/pdf',
     )
 
@@ -59,14 +57,12 @@ def run_with_openai() -> None:
 def run_with_anthropic() -> None:
     agent = Agent('anthropic:claude-3-5-sonnet-latest')
 
-    txt_url = MagicDocumentUrl(
+    txt_url = DocumentUrl(
         url='https://raw.githubusercontent.com/pydantic/pydantic/main/README.md',
-        filename='README.md',
         media_type='text/plain',
     )
-    pdf_url = MagicDocumentUrl(
+    pdf_url = DocumentUrl(
         url='https://arxiv.org/pdf/2403.05530.pdf',
-        filename='gemini-tech-report.pdf',
         media_type='application/pdf',
     )
 
diff --git a/pydantic_ai_slim/pydantic_ai/__init__.py b/pydantic_ai_slim/pydantic_ai/__init__.py
@@ -27,8 +27,6 @@
     BinaryContent,
     DocumentUrl,
     ImageUrl,
-    MagicBinaryContent,
-    MagicDocumentUrl,
     VideoUrl,
 )
 from .output import NativeOutput, PromptedOutput, StructuredDict, TextOutput, ToolOutput
@@ -62,8 +60,6 @@
     'VideoUrl',
     'DocumentUrl',
     'BinaryContent',
-    'MagicDocumentUrl',
-    'MagicBinaryContent',
     # tools
     'Tool',
     'ToolDefinition',
diff --git a/pydantic_ai_slim/pydantic_ai/messages.py b/pydantic_ai_slim/pydantic_ai/messages.py
@@ -529,76 +529,7 @@ def format(self) -> str:
     __repr__ = _utils.dataclasses_no_defaults_repr
 
 
-@dataclass(init=False, repr=False)
-class MagicDocumentUrl(DocumentUrl):
-    """A provider-agnostic document URL that may be transformed per adapter.
-
-    For OpenAI, text/plain documents may be converted to a plain text
-    `UserContent`.
-    """
-
-    filename: str | None = None
-    """Optional filename hint to use when converting to text."""
-
-    is_magic: Literal[True] = True
-    """Marker for serialization/filtering to indicate this is a magic part."""
-
-    def __init__(
-        self,
-        url: str,
-        *,
-        force_download: bool = False,
-        vendor_metadata: dict[str, Any] | None = None,
-        media_type: str | None = None,
-        filename: str | None = None,
-        identifier: str | None = None,
-        _media_type: str | None = None,
-    ) -> None:
-        super().__init__(
-            url=url,
-            force_download=force_download,
-            vendor_metadata=vendor_metadata,
-            media_type=media_type or _media_type,
-            identifier=identifier,
-        )
-        # Keep kind as 'document-url' for downstream OTEL/type expectations
-        self.filename = filename
-
-
-@dataclass(init=False, repr=False)
-class MagicBinaryContent(BinaryContent):
-    """A provider-agnostic binary content that may be transformed per adapter.
-
-    For OpenAI, text/plain content may be converted to a plain text
-    `UserContent`.
-    """
-
-    filename: str | None = None
-    """Optional filename hint to use when converting to text."""
-
-    is_magic: Literal[True] = True
-    """Marker for serialization/filtering to indicate this is a magic part."""
-
-    def __init__(
-        self,
-        data: bytes,
-        *,
-        media_type: AudioMediaType | ImageMediaType | DocumentMediaType | str,
-        filename: str | None = None,
-        identifier: str | None = None,
-        vendor_metadata: dict[str, Any] | None = None,
-    ) -> None:
-        super().__init__(
-            data=data,
-            media_type=media_type,
-            identifier=identifier,
-            vendor_metadata=vendor_metadata,
-        )
-        # Keep kind as 'binary' for downstream OTEL/type expectations
-        self.filename = filename
-
-
-MultiModalContent = ImageUrl | AudioUrl | DocumentUrl | VideoUrl | BinaryContent | MagicDocumentUrl | MagicBinaryContent
+MultiModalContent = ImageUrl | AudioUrl | DocumentUrl | VideoUrl | BinaryContent
 UserContent: TypeAlias = str | MultiModalContent
 
 
diff --git a/pydantic_ai_slim/pydantic_ai/models/openai.py b/pydantic_ai_slim/pydantic_ai/models/openai.py
@@ -26,8 +26,6 @@
     DocumentUrl,
     FinishReason,
     ImageUrl,
-    MagicBinaryContent,
-    MagicDocumentUrl,
     ModelMessage,
     ModelRequest,
     ModelResponse,
@@ -745,12 +743,7 @@ async def _map_user_prompt_items(items: Sequence[object]) -> list[ChatCompletion
     async def _map_single_item(item: object) -> list[ChatCompletionContentPartParam]:
         if isinstance(item, str):
             return [ChatCompletionContentPartTextParam(text=item, type='text')]
-        handled = await OpenAIChatModel._handle_magic_document(item)
-        if handled is not None:
-            return handled
-        handled = await OpenAIChatModel._handle_magic_binary(item)
-        if handled is not None:
-            return handled
+        # Magic* no longer used; logic ported to base handlers
         handled = OpenAIChatModel._handle_image_url(item)
         if handled is not None:
             return handled
@@ -768,55 +761,6 @@ async def _map_single_item(item: object) -> list[ChatCompletionContentPartParam]
         # Fallback: unknown type — return empty parts to avoid type-checker Never error
         return []
 
-    @staticmethod
-    async def _handle_magic_document(item: object) -> list[ChatCompletionContentPartParam] | None:
-        if not isinstance(item, MagicDocumentUrl):
-            return None
-        if OpenAIChatModel._is_text_like_media_type(item.media_type):
-            downloaded = await download_item(item, data_format='text', type_format='extension')
-            filename = item.filename or f'file.{downloaded["data_type"] or "txt"}'
-            inline = OpenAIChatModel._inline_file_block(filename, item.media_type, downloaded['data'])
-            return [ChatCompletionContentPartTextParam(text=inline, type='text')]
-        downloaded_item = await download_item(item, data_format='base64_uri', type_format='extension')
-        return [
-            File(
-                file=FileFile(
-                    file_data=downloaded_item['data'],
-                    filename=f'filename.{downloaded_item["data_type"]}',
-                ),
-                type='file',
-            )
-        ]
-
-    @staticmethod
-    async def _handle_magic_binary(item: object) -> list[ChatCompletionContentPartParam] | None:
-        if not isinstance(item, MagicBinaryContent):
-            return None
-        if OpenAIChatModel._is_text_like_media_type(item.media_type):
-            text = item.data.decode('utf-8')
-            filename = item.filename or 'file.txt'
-            inline = OpenAIChatModel._inline_file_block(filename, item.media_type, text)
-            return [ChatCompletionContentPartTextParam(text=inline, type='text')]
-        base64_encoded = base64.b64encode(item.data).decode('utf-8')
-        if item.is_image:
-            image_url = ImageURL(url=f'data:{item.media_type};base64,{base64_encoded}')
-            return [ChatCompletionContentPartImageParam(image_url=image_url, type='image_url')]
-        if item.is_audio:
-            assert item.format in ('wav', 'mp3')
-            audio = InputAudio(data=base64_encoded, format=item.format)
-            return [ChatCompletionContentPartInputAudioParam(input_audio=audio, type='input_audio')]
-        if item.is_document:
-            return [
-                File(
-                    file=FileFile(
-                        file_data=f'data:{item.media_type};base64,{base64_encoded}',
-                        filename=f'filename.{item.format}',
-                    ),
-                    type='file',
-                )
-            ]
-        raise RuntimeError(f'Unsupported binary content type: {item.media_type}')  # pragma: no cover
-
     @staticmethod
     def _handle_image_url(item: object) -> list[ChatCompletionContentPartParam] | None:
         if not isinstance(item, ImageUrl):
@@ -828,6 +772,27 @@ def _handle_image_url(item: object) -> list[ChatCompletionContentPartParam] | No
     async def _handle_binary_content(item: object) -> list[ChatCompletionContentPartParam] | None:
         if not isinstance(item, BinaryContent):
             return None
+        if OpenAIChatModel._is_text_like_media_type(item.media_type):
+            # Inline text-like binary content as a text block
+            text = item.data.decode('utf-8')
+            # Derive a sensible default filename from media type
+            media_type = item.media_type
+            if media_type == 'text/plain':
+                filename = 'file.txt'
+            elif media_type == 'text/csv':
+                filename = 'file.csv'
+            elif media_type == 'text/markdown':
+                filename = 'file.md'
+            elif media_type == 'application/json' or media_type.endswith('+json'):
+                filename = 'file.json'
+            elif media_type == 'application/xml' or media_type.endswith('+xml'):
+                filename = 'file.xml'
+            elif media_type in ('application/x-yaml', 'application/yaml', 'text/yaml'):
+                filename = 'file.yaml'
+            else:
+                filename = 'file.txt'
+            inline = OpenAIChatModel._inline_file_block(filename, media_type, text)
+            return [ChatCompletionContentPartTextParam(text=inline, type='text')]
         base64_encoded = base64.b64encode(item.data).decode('utf-8')
         if item.is_image:
             image_url = ImageURL(url=f'data:{item.media_type};base64,{base64_encoded}')
@@ -863,6 +828,11 @@ async def _handle_audio_url(item: object) -> list[ChatCompletionContentPartParam
     async def _handle_document_url(item: object) -> list[ChatCompletionContentPartParam] | None:
         if not isinstance(item, DocumentUrl):
             return None
+        if OpenAIChatModel._is_text_like_media_type(item.media_type):
+            downloaded_text = await download_item(item, data_format='text', type_format='extension')
+            filename = f'file.{downloaded_text["data_type"] or "txt"}'
+            inline = OpenAIChatModel._inline_file_block(filename, item.media_type, downloaded_text['data'])
+            return [ChatCompletionContentPartTextParam(text=inline, type='text')]
         downloaded_item = await download_item(item, data_format='base64_uri', type_format='extension')
         return [
             File(
diff --git a/tests/models/test_openai_textlike_mapping.py b/tests/models/test_openai_textlike_mapping.py
