sqlalchemy_async_session_factory.session¶

Async session factory and FastAPI dependency.

Extracted from GridFlow backend/src/database.py

Functions¶

`create_async_session_maker`(...)	Create async session maker for SQLAlchemy.
`create_session_dependency`(→ Callable[[], ...)	Create FastAPI dependency for database sessions.

Module Contents¶

sqlalchemy_async_session_factory.session.create_async_session_maker(engine: sqlalchemy.ext.asyncio.AsyncEngine, expire_on_commit: bool = False, **kwargs) → sqlalchemy.ext.asyncio.async_sessionmaker[sqlalchemy.ext.asyncio.AsyncSession]¶

Create async session maker for SQLAlchemy.

Args:

engine: AsyncEngine instance expire_on_commit: Expire objects on commit (default: False) **kwargs: Additional session maker configuration

Returns:

Configured async_sessionmaker instance

Example:

>>> from sqlalchemy.ext.asyncio import create_async_engine
>>> engine = create_async_engine("postgresql+asyncpg://...")
>>> SessionLocal = create_async_session_maker(engine)
>>>
>>> async with SessionLocal() as session:
...     # Use session
...     pass

sqlalchemy_async_session_factory.session.create_session_dependency(session_maker: sqlalchemy.ext.asyncio.async_sessionmaker[sqlalchemy.ext.asyncio.AsyncSession]) → Callable[[], AsyncGenerator[sqlalchemy.ext.asyncio.AsyncSession, None]]¶

Create FastAPI dependency for database sessions.

Returns a dependency function that provides database sessions to FastAPI endpoints with proper cleanup.

Args:

session_maker: async_sessionmaker instance

Returns:

Dependency function that yields AsyncSession

Example:

>>> from fastapi import FastAPI, Depends
>>>
>>> SessionLocal = create_async_session_maker(engine)
>>> get_db = create_session_dependency(SessionLocal)
>>>
>>> @app.get("/items")
>>> async def read_items(db: AsyncSession = Depends(get_db)):
...     # Use db session
...     pass