FastAPI sqlalchemy session per request handling

[theobouwman]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

HTTP test endpoint to test whether session is shared in the same request:

@router.get('/event/{event_id}')
@inject
async def testt(event_id: str, event_service: EventService = Depends(Provide[Container.event_service])):
    event1 = await event_service.get_event(event_id)
    event2 = await event_service.get_event(event_id)
    return BaseResponse[List[GetEventResponse]](data=[event1, event2])

EventService:

class EventService:
    def __init__(self, event_repository: EventRepository) -> None:
        self._event_repository: EventRepository = event_repository


    async def get_event(self, event_id: str) -> Event:
        event = await self._event_repository.get_by_id(event_id)
        if event is None:
            raise EntityNotFoundException

        return event

The EventRepository is just a subclass of the BaseRepository:

class BaseRepository(Generic[T]):
    _model: T

    def __init__(self, session_factory: Callable[..., AbstractContextManager[AsyncSession]]) -> None:
        self.session_factory = session_factory

    async def get_by_id(self, id: str) -> T:
        async with self.session_factory() as session:
            r = await session.execute(select(self._model).filter(self._model.id == id))
            return r.scalar_one_or_none()

Database manager:

class Database:

    def __init__(self, db_url: str) -> None:
        self._engine = create_async_engine(
            db_url,
            echo=get_config().QUERY_ECHO,
            echo_pool=get_config().ECHO_POOL,
            json_serializer=_custom_json_serializer,
            pool_pre_ping=True,
            pool_size=get_config().DB_POOL_SIZE,
        )
        self._async_session_factory = sessionmaker(
            bind=self._engine, 
            autocommit=False,
            autoflush=False,
            expire_on_commit=False,
            class_= AsyncSession
        )
        self._async_scoped_session = async_scoped_session(
            self._async_session_factory,
            scopefunc=asyncio.current_task,
        )

    def create_database(self) -> None:
        Base.metadata.create_all(self._engine)

    @asynccontextmanager
    async def session(self) -> Callable[..., AbstractContextManager[AsyncSession]]:
        session: AsyncSession = self._async_scoped_session()
        try:
            print('create session')
            yield session
        except Exception as e:
            await session.rollback()
            raise e
        finally:
            await session.close()

And our dependency injector:

class Container(containers.DeclarativeContainer):

    wiring_config = containers.WiringConfiguration(packages=[
        "api.routes",
        "tasks.routes",
        "common.observability"
    ])

    config = providers.Configuration()

    db = providers.Singleton(Database, db_url=get_config().DB_URL())

    event_repository = providers.Factory(
        EventRepository, session_factory=db.provided.session)

    event_service = providers.Factory(EventService,
                                      event_repository=event_repository,
                                      )

Description

We want to share the database session per request. Right now when we call the /event/{event_id} test route 2 sessions will be created, which is not what we want. We also see this in our traces on GCP on other routes (connect):

So what could be a solution to create a database session per request?

Operating System

Linux

Operating System Details

No response

FastAPI Version

0.103.2

Pydantic Version

2.4.2

Python Version

Python 3.11.5

Additional Context

No response

[ryanisn]

the 2 logged connect calls maybe just pure python calls?
per sqlalchemy the session should be reused unless you configured it not to
https://docs.sqlalchemy.org/en/20/orm/session_basics.html#closing

[mastizada]

A warning/tip in sqlalchemy docs:

SQLAlchemy generally does not recommend the “scoped” pattern for new development as it relies upon mutable global state that must also be explicitly torn down when work within the thread or task is complete. Particularly when using asyncio, it’s likely a better idea to pass the AsyncSession directly to the awaitable functions that need it.

We store the database session maker in the app state like this:

class DatabaseState:
    def __init__(self):
        self.engine = create_async_engine(
            current_config.SQLALCHEMY_DATABASE_URI,
            connect_args={
                "cursor_factory": AsyncClientCursor,
                "connect_timeout": 2,
                "keepalives_idle": 60,
            },
            # NullPool because we are using pgbouncer, if connecting directly to postgresql, use a pool
            poolclass=NullPool,
            echo=False,
        )

        # https://docs.sqlalchemy.org/en/20/orm/session_api.html#sqlalchemy.orm.Session.params.expire_on_commit
        self.session_maker: async_sessionmaker[AsyncSession] = async_sessionmaker(
            self.engine, expire_on_commit=False
        )

And then we have a dependency injection for database session:

async def get_db_session(request: Request) -> AsyncSession:
    async with request.app.state.database.session_maker() as session:
        try:
            yield session
        except Exception as e:
            logger.debug(f"Rolling back due to exception {e}.")
            await session.rollback()
            raise e
        finally:
            await session.close()

In our app, when we need the database session we use the dependency injection:

@router.get('/event/{event_id}')
async def test(event_id: str, event_service: EventService = Depends(EventService):
    event1 = await event_service.get_event(event_id)
    event2 = await event_service.get_event(event_id)
    return BaseResponse[List[GetEventResponse]](data=[event1, event2])

class EventService:
    def __init__(self, db_session: AsyncSession = Depends(get_db_session)) -> None:
        self.db_session = db_session

It works pretty well with multiple threads, the only issue is when pgbouncer goes to shutdown state (graceful shutdown) the connect step stays stuck until connection timeout. However, all active connection finish the operation successfully. We need a reconnect implementation, but we haven't had time to add that yet.