FastAPI Async SQLAlchemy

Drop-in async SQLAlchemy middleware for FastAPI. A request-scoped AsyncSession you reach through a single global db — no per-route dependency wiring, no manual session plumbing.

Get started View on GitHub

Why this middleware?

SQLAlchemy's AsyncSession is not safe to share across concurrent tasks, and FastAPI gives you a fresh request per coroutine. This middleware binds one session to each request context using a Python ContextVar, so db.session always resolves to the right session for the request you're in — whether you access it from a route, a service function, or a background helper.

from fastapi import FastAPI
from fastapi_async_sqlalchemy import SQLAlchemyMiddleware, db
from sqlalchemy import text

app = FastAPI()
app.add_middleware(
    SQLAlchemyMiddleware,
    db_url="postgresql+asyncpg://user:pass@localhost:5432/app",
)


@app.get("/ping")
async def ping():
    result = await db.session.execute(text("SELECT 1"))
    return {"db": result.scalar()}

No Depends(get_session), no passing the session down every call. Access the session anywhere in the request with db.session.

Features

• Request-scoped sessions

  ---

  A ContextVar-backed AsyncSession per request. Reach it from anywhere with db.session — no dependency injection boilerplate.

  Sessions & contexts

• Engine lifecycle done right

  ---

  Pass a db_url and the middleware owns and disposes the engine on shutdown; pass a custom_engine and you keep ownership.

  Engine lifecycle

• Pool-throttled concurrency

  ---

  Run many queries in parallel without exhausting the pool. db.gather() and db.connection() cap in-flight sessions at max_concurrent.

  Concurrent queries

• Multiple databases

  ---

  Build independent middleware/proxy pairs with create_middleware_and_session_proxy() — one per database.

  Multiple databases

• Streaming-aware

  ---

  Clear rules for StreamingResponse so the session lifetime belongs to the body, not a closed request transaction.

  Streaming responses

• Typed & SQLModel-ready

  ---

  Ships py.typed, a DBSessionMeta Protocol for full autocomplete, and works transparently with sqlmodel.

  Type hints

Installation

pip install fastapi-async-sqlalchemy

Requires Python 3.12+, starlette>=0.40, and SQLAlchemy>=2.0. Add the async driver for your database (asyncpg, aiomysql, aiosqlite, …).