Bug: Litestar `SQLAlchemyAsyncConfig.create_session_maker()` does not register FileObject session listeners

[hasansezertasan]

Description

When using FileObject / StoredObject with the Litestar extension, files are never uploaded to storage. The file metadata is persisted to the database, but the actual file content is silently discarded because the AsyncFileObjectListener (before_flush, after_commit, after_rollback) is never registered on the session.

The root cause is that the Litestar-specific SQLAlchemyAsyncConfig.create_session_maker() overrides the base class method without calling super(), skipping all listener registration that the base class performs.

Base class (registers listeners):

# config/asyncio.py
def create_session_maker(self) -> Callable[[], AsyncSession]:
    ...
    sync_maker = sync_sessionmaker()
    if self.enable_file_object_listener:
        event.listen(sync_maker, "before_flush", AsyncFileObjectListener.before_flush)
        event.listen(sync_maker, "after_commit", AsyncFileObjectListener.after_commit)
        event.listen(sync_maker, "after_rollback", AsyncFileObjectListener.after_rollback)
    session_maker.configure(sync_session_class=sync_maker)
    return session_maker

Litestar override (skips listeners):

# extensions/litestar/.../asyncio.py
def create_session_maker(self) -> Callable[[], AsyncSession]:
    if self.session_maker:
        return self.session_maker
    session_kws = self.session_config_dict
    if session_kws.get("bind") is None:
        session_kws["bind"] = self.get_engine()
    return self.session_maker_class(**session_kws)

Workaround: manually register the listeners before creating the Litestar app:

from sqlalchemy import event
from sqlalchemy.orm import sessionmaker as sync_sessionmaker
from advanced_alchemy._listeners import AsyncFileObjectListener

sync_maker = sync_sessionmaker()
event.listen(sync_maker, "before_flush", AsyncFileObjectListener.before_flush)
event.listen(sync_maker, "after_commit", AsyncFileObjectListener.after_commit)
event.listen(sync_maker, "after_rollback", AsyncFileObjectListener.after_rollback)
alchemy_config.session_maker = alchemy_config.session_maker_class(
    bind=alchemy_config.get_engine(),
    sync_session_class=sync_maker,
    **{k: v for k, v in alchemy_config.session_config_dict.items() if k not in ("bind", "sync_session_class")},
)

Suggested fix: The Litestar SQLAlchemyAsyncConfig.create_session_maker() should delegate to super().create_session_maker() (or replicate the listener registration logic) so that AsyncFileObjectListener events are attached when enable_file_object_listener is True.

MCVE

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "advanced-alchemy[obstore]==1.9.3",
#     "litestar==2.21.1",
#     "aiosqlite==0.22.1",
#     "pydantic",
#     "uvicorn",
# ]
# ///
from __future__ import annotations

import shutil
from pathlib import Path
from typing import Annotated, Any
from uuid import UUID

import uvicorn
from advanced_alchemy.extensions.litestar import (
    AsyncSessionConfig,
    SQLAlchemyAsyncConfig,
    SQLAlchemyPlugin,
    base,
    providers,
    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
from litestar import Controller, Litestar, get, post
from litestar.datastructures import UploadFile
from litestar.enums import RequestEncodingType
from litestar.params import Body
from pydantic import BaseModel
from sqlalchemy.orm import Mapped, mapped_column

UPLOAD_DIR = Path("./test_uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

storages.register_backend(
    ObstoreBackend(fs=UPLOAD_DIR.resolve().as_uri() + "/", key="local"),
)


class DocumentModel(base.UUIDBase):
    __tablename__ = "document"
    name: Mapped[str]
    file: Mapped[FileObject] = mapped_column(StoredObject(backend="local"))


class Document(BaseModel):
    id: UUID | None
    name: str


class CreateDocument(BaseModel):
    model_config = {"arbitrary_types_allowed": True}
    name: str
    file: UploadFile | None = None


class DocumentService(service.SQLAlchemyAsyncRepositoryService[DocumentModel]):
    class Repo(repository.SQLAlchemyAsyncRepository[DocumentModel]):
        model_type = DocumentModel

    repository_type = Repo


class DocumentController(Controller):
    path = "/documents"
    dependencies = providers.create_service_dependencies(
        DocumentService, "documents_service", load=[DocumentModel.file],
    )

    @post(path="/")
    async def create_document(
        self,
        data: Annotated[CreateDocument, Body(media_type=RequestEncodingType.MULTI_PART)],
        documents_service: DocumentService,
    ) -> Document:
        obj = await documents_service.create(
            DocumentModel(
                name=data.name,
                file=FileObject(
                    backend="local",
                    filename=data.file.filename or "uploaded_file",
                    content_type=data.file.content_type,
                    content=await data.file.read(),
                )
                if data.file
                else None,
            )
        )
        return documents_service.to_schema(obj, schema_type=Document)

    @get(path="/check/{document_id:uuid}")
    async def check_file_exists(
        self,
        documents_service: DocumentService,
        document_id: UUID,
    ) -> dict[str, Any]:
        obj = await documents_service.get(document_id)
        file_path = UPLOAD_DIR / obj.file.path if obj.file else None
        return {
            "document_id": str(document_id),
            "file_metadata_in_db": obj.file.to_dict() if obj.file else None,
            "file_exists_on_disk": file_path.exists() if file_path else False,
        }


alchemy_config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///repro.sqlite3",
    session_config=AsyncSessionConfig(expire_on_commit=False),
    before_send_handler="autocommit",
    create_all=True,
)

app = Litestar(
    route_handlers=[DocumentController],
    plugins=[SQLAlchemyPlugin(config=alchemy_config)],
    debug=True,
)

if __name__ == "__main__":
    Path("repro.sqlite3").unlink(missing_ok=True)
    shutil.rmtree(UPLOAD_DIR, ignore_errors=True)
    UPLOAD_DIR.mkdir(exist_ok=True)
    uvicorn.run(app, host="0.0.0.0", port=8000)

Steps to reproduce

1. Save the MCVE as issue_repro.py and run with uv run issue_repro.py
2. Upload a file: curl -s -X POST http://localhost:8000/documents -F 'name=test' -F 'file=@any_file.txt;type=text/plain'
3. Check if the file was written: curl -s http://localhost:8000/documents/check/<id> | python -m json.tool
4. Observe "file_exists_on_disk": false

Logs

No errors or warnings are emitted. The upload failure is completely silent because AsyncFileObjectListener is never registered, so _pending_source_content on the FileObject is simply ignored during the session lifecycle.

Package Version

• advanced-alchemy==1.9.3
• litestar==2.21.1
• obstore==0.9.2

Platform

• Mac

---

This issue was created by @hasansezertasan with the help of Claude Code.