# Async SQLAlchemy Session Management

!!! info "Primary sources"
    - [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
    - [SQLAlchemy session basics](https://docs.sqlalchemy.org/en/21/orm/session_basics.html)
    - [FastAPI dependencies with yield](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/)

??? abstract "Decision metadata"
    - Status: adopted
    - Decision level: mandatory
    - Applies to: api-runtime, workers, tests
    - Last reviewed: 2026-06-17

---

## Purpose

Define one canonical session model for FastAPI + SQLAlchemy asyncio:

- configure one shared session factory,
- create one AsyncSession per request or per unit-of-work,
- never share one AsyncSession across concurrent tasks.

---

## Scope and Non-Goals

- In scope: session factory creation, FastAPI dependency wiring, request/task scoping, transaction demarcation.
- Out of scope: ORM model design, query optimization strategy, schema migration tooling.

---

## Rules

- Create `async_sessionmaker` once from app-owned AsyncEngine.
- Use a fresh AsyncSession for each request or explicit unit-of-work.
- Do not share AsyncSession across `asyncio.gather()` or parallel tasks.
- Prefer direct dependency injection over global scoped-session patterns in new code.
- Use explicit transaction boundaries (`async with session.begin():`) for writes.

---

## Canonical FastAPI Dependency Pattern

```python
from collections.abc import AsyncIterator

from fastapi import Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker


def get_session_factory(request: Request) -> async_sessionmaker[AsyncSession]:
    return request.app.state.session_factory


async def get_db_session(
    session_factory: async_sessionmaker[AsyncSession] = Depends(get_session_factory),
) -> AsyncIterator[AsyncSession]:
    async with session_factory() as session:
    yield session
```

Route usage:

```python
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter()


@router.post("/items")
async def create_item(session: AsyncSession = Depends(get_db_session)) -> dict:
    async with session.begin():
    # write operations here
    ...
    return {"status": "ok"}
```

---

## Configuration Guidance

Typical session factory setup:

```python
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)
```

Notes:

- `expire_on_commit=False` is commonly preferred in asyncio applications to reduce accidental post-commit reload behavior.
- `AsyncSession.refresh()` is preferred over broad expiration patterns when state refresh is needed.

---

## Concurrency Rules

- One session per concurrent task.
- If work fans out into parallel tasks, each task receives its own AsyncSession.
- Pass sessions explicitly to service functions; avoid mutable global session state.

---

## Anti-Patterns

- A singleton/global AsyncSession reused across requests.
- Sharing one AsyncSession across parallel tasks.
- Hidden session creation in lower repository helpers with no caller control.
- Mixing commit/rollback ownership across layers without a declared boundary.

---

## Operational Checks

- Exactly one `async_sessionmaker` is registered in app lifecycle.
- Request handlers receive sessions from one canonical dependency.
- No code path creates AsyncSession in module import side effects.
- Background jobs and API handlers each create task-local sessions.

---

## Testing Checks

- Dependency override exists for test session factory.
- Rollback behavior is verified for failed write units.
- Parallel-task tests verify no shared AsyncSession instances.
- Lifespan tests confirm session factory is initialized and teardown-safe.

---

## Migration Notes

- If current code uses global/shared sessions, fix scope first before refactoring query style.
- If legacy sync patterns are present, keep session boundary rules stable while migrating incrementally.