Async Databases

Q: SQLAlchemy 2 pattern?

create_async_engine, async_sessionmaker, async with session.begin().

Q: SQLite async?

aiosqlite for dev/small apps - different concurrency limits than Postgres.

Q: Pydantic with rows?

model_validate on dict rows from mapping().all() - validate at boundary.

Async database access keeps the event loop free while waiting on the server. asyncpg (PostgreSQL), SQLAlchemy 2 async, and driver-specific pools integrate with FastAPI lifespan and per-request sessions.

Recipe

# Conceptual SQLAlchemy 2 async + asyncpg
# engine = create_async_engine("postgresql+asyncpg://...")
# async with AsyncSession(engine) as session:
#     result = await session.execute(select(User).limit(1))

When to reach for this:

FastAPI routes hitting PostgreSQL under concurrency
Long-poll or websocket apps sharing DB pool
Microservices with I/O-bound query workloads
Replacing sync SQLAlchemy in async def routes
Connection pool tuning for container limits

Working Example

Simulated asyncpg-style flow with asyncio stand-ins for runnable demo without live DB.

import asyncio
from dataclasses import dataclass, field
 
@dataclass
class FakePool:
    _connections: int = 0
    max_size: int = 5
 
    async def acquire(self) -> "FakeConn":
        await asyncio.sleep(0.01)
        self._connections += 1
        return FakeConn(self)
 
    async def release(self, conn: "FakeConn") -> None:
        await asyncio.sleep(0)
        self._connections -= 1
 
@dataclass
class FakeConn:
    pool: FakePool
 
    async def fetchval(self, query: str) -> int:
        await asyncio.sleep(0.02)
        return 42
 
class DB:
    def __init__(self, pool: FakePool) -> None:
        self._pool = pool
 
    async def user_count(self) -> int:
        conn = await self._pool.acquire()
        try:
            return await conn.fetchval("SELECT COUNT(*) FROM users")
        finally:
            await self._pool.release(conn)
 
async def main() -> None:
    pool = FakePool()
    db = DB(pool)
    counts = await asyncio.gather(*(db.user_count() for _ in range(3)))
    print(counts, "pool conns", pool._connections)
 
asyncio.run(main())

What this demonstrates:

Acquire/release pattern mirrors real pools
finally ensures connection returns to pool
Concurrent queries overlap waits on one loop
Track pool size to avoid exhausting DB max connections

Deep Dive

Stack Options

Layer	PostgreSQL
Driver	asyncpg
ORM	SQLAlchemy 2 `AsyncSession`
Migrations	Alembic (sync engine) separate job

Session Scope

Prefer one session per request
Commit in successful request path; rollback on exception
Do not share session across concurrent tasks

Gotchas

Sync session in async route - blocks loop. Fix: AsyncSession + await execute.
Pool per request - connection storm. Fix: app-scoped engine/pool in lifespan.
Missing await on execute - returns coroutine, silent logic bugs. Fix: pyright/ruff ASYNC rules.
N+1 queries in async - still bad with parallelism. Fix: eager loading, batch queries.
Transactions spanning awaits to external APIs - long locks. Fix: short DB transactions only around DB work.

Alternatives

Alternative	Use When	Don't Use When
Sync ORM + thread pool	Gradual migration	Greenfield async API
Raw asyncpg	Max perf, SQL control	Heavy ORM needs
DynamoDB async SDK	AWS NoSQL	Relational needs

FAQs

asyncpg vs psycopg3 async?

Both viable; SQLAlchemy abstracts driver - pick per ops familiarity and feature needs.

Pool size formula?

(expected_concurrent_requests * avg_query_time) / target_latency bounded by DB max_connections.

SQLAlchemy 2 pattern?

create_async_engine, async_sessionmaker, async with session.begin().

Migrations async?

Alembic typically runs sync migration job in CI - not in request hot path.

SQLite async?

aiosqlite for dev/small apps - different concurrency limits than Postgres.

Django async ORM?

Improving - check Django 5.2 docs for async query support on your models.

How to test?

Testcontainers Postgres or sqlite+aiosqlite with pytest-asyncio fixtures.

Connection leaks?

Always context-manage sessions; monitor pool checked-out count metric.

Read replicas?

Separate engines/pools for read vs write routing in app layer.

Pydantic with rows?

model_validate on dict rows from mapping().all() - validate at boundary.

Async Context Managers & Iterators - session scopes
Structured Concurrency - request task scope
Sync/Async Bridging - avoid sync ORM
FastAPI async routes - framework wiring

Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.

Recipe

# Conceptual SQLAlchemy 2 async + asyncpg
# engine = create_async_engine("postgresql+asyncpg://...")
# async with AsyncSession(engine) as session:
#     result = await session.execute(select(User).limit(1))

When to reach for this:

FastAPI routes hitting PostgreSQL under concurrency
Long-poll or websocket apps sharing DB pool
Microservices with I/O-bound query workloads
Replacing sync SQLAlchemy in async def routes
Connection pool tuning for container limits

Working Example

Simulated asyncpg-style flow with asyncio stand-ins for runnable demo without live DB.

import asyncio
from dataclasses import dataclass, field
 
@dataclass
class FakePool:
    _connections: int = 0
    max_size: int = 5
 
    async def acquire(self) -> "FakeConn":
        await asyncio.sleep(0.01)
        self._connections += 1
        return FakeConn(self)
 
    async def release(self, conn: "FakeConn") -> None:
        await asyncio.sleep(0)
        self._connections -= 1
 
@dataclass
class FakeConn:
    pool: FakePool
 
    async def fetchval(self, query: str) -> int:
        await asyncio.sleep(0.02)
        return 42
 
class DB:
    def __init__(self, pool: FakePool) -> None:
        self._pool = pool
 
    async def user_count(self) -> int:
        conn = await self._pool.acquire()
        try:
            return await conn.fetchval("SELECT COUNT(*) FROM users")
        finally:
            await self._pool.release(conn)
 
async def main() -> None:
    pool = FakePool()
    db = DB(pool)
    counts = await asyncio.gather(*(db.user_count() for _ in range(3)))
    print(counts, "pool conns", pool._connections)
 
asyncio.run(main())

What this demonstrates:

Acquire/release pattern mirrors real pools
finally ensures connection returns to pool
Concurrent queries overlap waits on one loop
Track pool size to avoid exhausting DB max connections

Deep Dive

Stack Options

Layer	PostgreSQL
Driver	asyncpg
ORM	SQLAlchemy 2 `AsyncSession`
Migrations	Alembic (sync engine) separate job

Session Scope

Prefer one session per request
Commit in successful request path; rollback on exception
Do not share session across concurrent tasks

Gotchas

Sync session in async route - blocks loop. Fix: AsyncSession + await execute.
Pool per request - connection storm. Fix: app-scoped engine/pool in lifespan.
Missing await on execute - returns coroutine, silent logic bugs. Fix: pyright/ruff ASYNC rules.
N+1 queries in async - still bad with parallelism. Fix: eager loading, batch queries.
Transactions spanning awaits to external APIs - long locks. Fix: short DB transactions only around DB work.

Alternatives

Alternative	Use When	Don't Use When
Sync ORM + thread pool	Gradual migration	Greenfield async API
Raw asyncpg	Max perf, SQL control	Heavy ORM needs
DynamoDB async SDK	AWS NoSQL	Relational needs

FAQs

asyncpg vs psycopg3 async?

Both viable; SQLAlchemy abstracts driver - pick per ops familiarity and feature needs.

Pool size formula?

(expected_concurrent_requests * avg_query_time) / target_latency bounded by DB max_connections.

SQLAlchemy 2 pattern?

create_async_engine, async_sessionmaker, async with session.begin().

Migrations async?

Alembic typically runs sync migration job in CI - not in request hot path.

SQLite async?

aiosqlite for dev/small apps - different concurrency limits than Postgres.

Django async ORM?

Improving - check Django 5.2 docs for async query support on your models.

How to test?

Testcontainers Postgres or sqlite+aiosqlite with pytest-asyncio fixtures.

Connection leaks?

Always context-manage sessions; monitor pool checked-out count metric.

Read replicas?

Separate engines/pools for read vs write routing in app layer.

Pydantic with rows?

model_validate on dict rows from mapping().all() - validate at boundary.

Async Context Managers & Iterators - session scopes
Structured Concurrency - request task scope
Sync/Async Bridging - avoid sync ORM
FastAPI async routes - framework wiring

Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.

Async Databases

Search Documentation

Async Databases