fix: FileObject native Pydantic Core integration (#458)

File object will now serialize properly in pydantic.

More complete FastAPI examples added.

Author: cofin

.pre-commit-config.yaml

@@ -22,7 +22,7 @@ repos:
       - id: unasyncd
         additional_dependencies: ["ruff"]
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: "v0.11.6"
+    rev: "v0.11.7"
     hooks:
       # Run the linter.
       - id: ruff

advanced_alchemy/types/file_object/_typing.py

@@ -0,0 +1,44 @@
+"""Internal typing helpers for file_object, handling optional Pydantic integration."""
+
+from typing import Any, Protocol, TypeVar, runtime_checkable
+
+# Define a generic type variable for CoreSchema placeholder if needed
+CoreSchemaT = TypeVar("CoreSchemaT")
+
+try:
+    # Attempt to import real Pydantic components
+    from pydantic import GetCoreSchemaHandler  # pyright: ignore
+    from pydantic_core import core_schema  # pyright: ignore
+
+    PYDANTIC_INSTALLED = True
+
+except ImportError:
+    PYDANTIC_INSTALLED = False  # pyright: ignore
+
+    @runtime_checkable
+    class GetCoreSchemaHandler(Protocol):  # type: ignore[no-redef]
+        """Placeholder for Pydantic's GetCoreSchemaHandler."""
+
+        def __call__(self, source_type: Any) -> Any: ...
+
+        def __getattr__(self, item: str) -> Any:  # Allow arbitrary attribute access
+            return Any
+
+    # Define a placeholder for core_schema module
+    class CoreSchemaModulePlaceholder:
+        """Placeholder for pydantic_core.core_schema module."""
+
+        # Define placeholder types/functions used in FileObject.__get_pydantic_core_schema__
+        CoreSchema = Any  # Placeholder for the CoreSchema type itself
+
+        def __getattr__(self, name: str) -> Any:
+            """Return a dummy function/type for any requested attribute."""
+
+            def dummy_schema_func(*args: Any, **kwargs: Any) -> Any:  # noqa: ARG001
+                return Any
+
+            return dummy_schema_func
+
+    core_schema = CoreSchemaModulePlaceholder()  # type: ignore[assignment]
+
+__all__ = ("GetCoreSchemaHandler", "core_schema")

advanced_alchemy/types/file_object/file.py

@@ -1,4 +1,3 @@
-# ruff: noqa: PLR0904, PLR6301
 """Generic unified storage protocol compatible with multiple backend implementations."""
 
 import mimetypes
@@ -8,6 +7,8 @@
 from sqlalchemy.ext.mutable import MutableList
 from typing_extensions import TypeAlias
 
+from advanced_alchemy.exceptions import MissingDependencyError
+from advanced_alchemy.types.file_object._typing import PYDANTIC_INSTALLED, GetCoreSchemaHandler, core_schema
 from advanced_alchemy.types.file_object.base import AsyncDataLike, DataLike, StorageBackend
 from advanced_alchemy.types.file_object.registry import storages
 
@@ -409,5 +410,93 @@ async def save_async(
 
         return updated_self
 
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls,
+        source_type: Any,
+        handler: "GetCoreSchemaHandler",  # Use imported GetCoreSchemaHandler
+    ) -> "core_schema.CoreSchema":  # Use imported core_schema
+        """Get the Pydantic core schema for FileObject.
+
+        This method defines how Pydantic should validate and serialize FileObject instances.
+        It creates a schema that validates dictionaries with the required fields and
+        converts them to FileObject instances.
+
+        Raises:
+            MissingDependencyError: If Pydantic is not installed when this method is called.
+
+        Args:
+            source_type: The source type (FileObject)
+            handler: The Pydantic schema handler
+
+        Returns:
+            A Pydantic core schema for FileObject
+        """
+        if not PYDANTIC_INSTALLED:
+            raise MissingDependencyError(package="pydantic")
+
+        def validate_from_dict(data: dict[str, Any]) -> "FileObject":
+            # We expect a dictionary derived from to_dict()
+            # We need to resolve the backend string back to an instance if needed
+            backend_input = data.get("backend")
+            if backend_input is None:
+                msg = "backend is required"
+                raise TypeError(msg)
+            key = backend_input if isinstance(backend_input, str) else backend_input.key
+            return cls(
+                backend=key,
+                filename=data["filename"],
+                to_filename=data.get("to_filename"),
+                content_type=data.get("content_type"),
+                size=data.get("size"),
+                last_modified=data.get("last_modified"),
+                checksum=data.get("checksum"),
+                etag=data.get("etag"),
+                version_id=data.get("version_id"),
+                metadata=data.get("metadata"),
+            )
+
+        typed_dict_schema = core_schema.typed_dict_schema(
+            {
+                "filename": core_schema.typed_dict_field(core_schema.str_schema()),
+                "backend": core_schema.typed_dict_field(core_schema.str_schema()),
+                "to_filename": core_schema.typed_dict_field(core_schema.str_schema(), required=False),
+                "content_type": core_schema.typed_dict_field(core_schema.str_schema(), required=False),
+                "size": core_schema.typed_dict_field(core_schema.int_schema(), required=False),
+                "last_modified": core_schema.typed_dict_field(core_schema.float_schema(), required=False),
+                "checksum": core_schema.typed_dict_field(core_schema.str_schema(), required=False),
+                "etag": core_schema.typed_dict_field(core_schema.str_schema(), required=False),
+                "version_id": core_schema.typed_dict_field(core_schema.str_schema(), required=False),
+                "metadata": core_schema.typed_dict_field(
+                    core_schema.nullable_schema(
+                        core_schema.dict_schema(core_schema.str_schema(), core_schema.any_schema())
+                    ),
+                    required=False,
+                ),
+            }
+        )
+
+        validation_schema = core_schema.union_schema(
+            [
+                core_schema.is_instance_schema(cls),
+                core_schema.chain_schema(
+                    [
+                        typed_dict_schema,
+                        core_schema.no_info_plain_validator_function(validate_from_dict),
+                    ]
+                ),
+            ]
+        )
+
+        return core_schema.json_or_python_schema(
+            json_schema=validation_schema,
+            python_schema=validation_schema,
+            serialization=core_schema.plain_serializer_function_ser_schema(
+                lambda instance: instance.to_dict(),  # pyright: ignore
+                info_arg=False,
+                return_schema=typed_dict_schema,
+            ),  # pyright: ignore
+        )
+
 
 FileObjectList: TypeAlias = MutableList[FileObject]

examples/fastapi/fastapi_fileobject.py

@@ -0,0 +1,171 @@
+# /// script
+# dependencies = [
+#   "advanced_alchemy[obstore,uuid]",
+#   "aiosqlite",
+#   "fastapi[standard]",
+#   "orjson"
+#   "obstore"
+# ]
+# ///
+from typing import Annotated, Any, Optional, Union
+from uuid import UUID
+
+import uvicorn
+from fastapi import APIRouter, Depends, FastAPI, File, Form, UploadFile
+from pydantic import BaseModel, Field, computed_field
+from sqlalchemy.orm import Mapped, mapped_column
+
+from advanced_alchemy.extensions.fastapi import (
+    AdvancedAlchemy,
+    AsyncSessionConfig,
+    SQLAlchemyAsyncConfig,
+    base,
+    filters,
+    repository,
+    service,
+)
+from advanced_alchemy.types import FileObject, storages
+from advanced_alchemy.types.file_object.backends.obstore import ObstoreBackend
+from advanced_alchemy.types.file_object.data_type import StoredObject
+
+sqlalchemy_config = SQLAlchemyAsyncConfig(
+    connection_string="sqlite+aiosqlite:///test.sqlite",
+    session_config=AsyncSessionConfig(expire_on_commit=False),
+    commit_mode="autocommit",
+    create_all=True,
+)
+app = FastAPI()
+alchemy = AdvancedAlchemy(config=sqlalchemy_config, app=app)
+document_router = APIRouter()
+s3_backend = ObstoreBackend(
+    key="local",
+    fs="s3://static-files/",
+    aws_endpoint="http://localhost:9000",
+    aws_access_key_id="minioadmin",
+    aws_secret_access_key="minioadmin",  # noqa: S106
+)
+storages.register_backend(s3_backend)
+
+
+class DocumentModel(base.UUIDBase):
+    # we can optionally provide the table name instead of auto-generating it
+    __tablename__ = "document"
+    name: Mapped[str]
+    file: Mapped[FileObject] = mapped_column(StoredObject(backend="local"))
+
+
+class DocumentService(service.SQLAlchemyAsyncRepositoryService[DocumentModel]):
+    """Author repository."""
+
+    class Repo(repository.SQLAlchemyAsyncRepository[DocumentModel]):
+        """Author repository."""
+
+        model_type = DocumentModel
+
+    repository_type = Repo
+
+
+# Pydantic Models
+
+
+class Document(BaseModel):
+    id: Optional[UUID]
+    name: str
+    file: Optional[FileObject] = Field(default=None, exclude=True)
+
+    @computed_field
+    def file_url(self) -> Optional[Union[str, list[str]]]:
+        if self.file is None:
+            return None
+        return self.file.sign()
+
+
+@document_router.get(path="/documents", response_model=service.OffsetPagination[Document])
+async def list_documents(
+    documents_service: Annotated[
+        DocumentService, Depends(alchemy.provide_service(DocumentService, load=[DocumentModel.file]))
+    ],
+    filters: Annotated[
+        list[filters.FilterTypes],
+        Depends(
+            alchemy.provide_filters(
+                {
+                    "id_filter": UUID,
+                    "pagination_type": "limit_offset",
+                    "search": "name",
+                    "search_ignore_case": True,
+                }
+            )
+        ),
+    ],
+) -> service.OffsetPagination[Document]:
+    results, total = await documents_service.list_and_count(*filters)
+    return documents_service.to_schema(results, total, filters=filters, schema_type=Document)
+
+
+@document_router.post(path="/documents")
+async def create_document(
+    documents_service: Annotated[DocumentService, Depends(alchemy.provide_service(DocumentService))],
+    name: Annotated[str, Form()],
+    file: Annotated[Optional[UploadFile], File()] = None,
+) -> Document:
+    obj = await documents_service.create(
+        DocumentModel(
+            name=name,
+            file=FileObject(
+                backend="local",
+                filename=file.filename or "uploaded_file",
+                content_type=file.content_type,
+                content=await file.read(),
+            )
+            if file
+            else None,
+        )
+    )
+    return documents_service.to_schema(obj, schema_type=Document)
+
+
+@document_router.get(path="/documents/{document_id}")
+async def get_document(
+    documents_service: Annotated[DocumentService, Depends(alchemy.provide_service(DocumentService))],
+    document_id: UUID,
+) -> Document:
+    obj = await documents_service.get(document_id)
+    return documents_service.to_schema(obj, schema_type=Document)
+
+
+@document_router.patch(path="/documents/{document_id}")
+async def update_document(
+    documents_service: Annotated[DocumentService, Depends(alchemy.provide_service(DocumentService))],
+    document_id: UUID,
+    name: Annotated[Optional[str], Form()] = None,
+    file: Annotated[Optional[UploadFile], File()] = None,
+) -> Document:
+    update_data: dict[str, Any] = {}
+    if name is not None:
+        update_data["name"] = name
+    if file is not None:
+        update_data["file"] = FileObject(
+            backend="local",
+            filename=file.filename or "uploaded_file",
+            content_type=file.content_type,
+            content=await file.read(),
+        )
+
+    obj = await documents_service.update(update_data, item_id=document_id)
+    return documents_service.to_schema(obj, schema_type=Document)
+
+
+@document_router.delete(path="/documents/{document_id}")
+async def delete_document(
+    documents_service: Annotated[DocumentService, Depends(alchemy.provide_service(DocumentService))],
+    document_id: UUID,
+) -> None:
+    _ = await documents_service.delete(document_id)
+
+
+app.include_router(document_router)
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)  # noqa: S104