diff --git a/pydantic_ai_slim/pydantic_ai/messages.py b/pydantic_ai_slim/pydantic_ai/messages.py index 1f3b5cd6e5..9019b81931 100644 --- a/pydantic_ai_slim/pydantic_ai/messages.py +++ b/pydantic_ai_slim/pydantic_ai/messages.py @@ -7,6 +7,8 @@ from dataclasses import KW_ONLY, dataclass, field, replace from datetime import datetime from mimetypes import guess_type +from os import PathLike +from pathlib import Path from typing import TYPE_CHECKING, Annotated, Any, Literal, TypeAlias, cast, overload import pydantic @@ -530,6 +532,25 @@ def from_data_uri(cls, data_uri: str) -> BinaryContent: media_type, data = data_uri[len(prefix) :].split(';base64,', 1) return cls.narrow_type(cls(data=base64.b64decode(data), media_type=media_type)) + @classmethod + def from_path(cls, path: PathLike[str]) -> BinaryContent: + """Create a `BinaryContent` from a path. + + Defaults to 'application/octet-stream' if the media type cannot be inferred. + + Raises: + FileNotFoundError: if the file does not exist. + PermissionError: if the file cannot be read. + """ + path = Path(path) + if not path.exists(): + raise FileNotFoundError(f'File not found: {path}') + media_type, _ = guess_type(path) + if media_type is None: + media_type = 'application/octet-stream' + + return cls.narrow_type(cls(data=path.read_bytes(), media_type=media_type)) + @pydantic.computed_field @property def identifier(self) -> str: diff --git a/tests/test_messages.py b/tests/test_messages.py index 627be2c928..943d68fe8c 100644 --- a/tests/test_messages.py +++ b/tests/test_messages.py @@ -1,5 +1,6 @@ import sys from datetime import datetime, timezone +from pathlib import Path import pytest from inline_snapshot import snapshot @@ -612,3 +613,37 @@ def test_binary_content_validation_with_optional_identifier(): 'identifier': 'foo', } ) + + +def test_binary_content_from_path(tmp_path: Path): + # test normal file + test_xml_file = tmp_path / 'test.xml' + test_xml_file.write_text('about trains', encoding='utf-8') + binary_content = BinaryContent.from_path(test_xml_file) + assert binary_content == snapshot(BinaryContent(data=b'about trains', media_type='application/xml')) + + # test non-existent file + non_existent_file = tmp_path / 'non-existent.txt' + with pytest.raises(FileNotFoundError, match='File not found:'): + BinaryContent.from_path(non_existent_file) + + # test file with unknown media type + test_unknown_file = tmp_path / 'test.unknownext' + test_unknown_file.write_text('some content', encoding='utf-8') + binary_content = BinaryContent.from_path(test_unknown_file) + assert binary_content == snapshot(BinaryContent(data=b'some content', media_type='application/octet-stream')) + + # test string path + test_txt_file = tmp_path / 'test.txt' + test_txt_file.write_text('just some text', encoding='utf-8') + string_path = test_txt_file.as_posix() + binary_content = BinaryContent.from_path(string_path) # pyright: ignore[reportArgumentType] + assert binary_content == snapshot(BinaryContent(data=b'just some text', media_type='text/plain')) + + # test image file + test_jpg_file = tmp_path / 'test.jpg' + test_jpg_file.write_bytes(b'\xff\xd8\xff\xe0' + b'0' * 100) # minimal JPEG header + padding + binary_content = BinaryContent.from_path(test_jpg_file) + assert binary_content == snapshot( + BinaryImage(data=b'\xff\xd8\xff\xe0' + b'0' * 100, media_type='image/jpeg', _identifier='bc8d49') + )