Async SQLAlchemy and SQLModel
Scenario
Your FastAPI app can handle 10k req/sec in a synthetic benchmark, but the moment you connect it to Postgres, it plummets to 200 req/sec and throws timeout errors everywhere.
Mental model
Async DB drivers are like placing an order at a busy kitchen: you drop the ticket and step back to do other work, instead of standing at the pass blocking the only waiter until your food is ready.
FastAPI runs on a single-threaded event loop. If you use a synchronous driver, that single thread stops entirely while waiting for the database. An asynchronous driver (like asyncpg) pauses only the specific request, allowing the event loop to serve thousands of other requests in the meantime.
Deep dive
FastAPI is built around an event loop (asyncio). It achieves high concurrency using a single thread that switches tasks whenever it hits an I/O pause (like a database query). If you use a synchronous database driver (like psycopg2), the driver blocks the entire thread while waiting for the database to reply. The event loop freezes. No other requests can be served.
Synchronous drivers freeze the event loop, tanking performance.
To fix this, we use an asynchronous driver (like asyncpg for PostgreSQL). It returns control to the event loop immediately when it sends the query over the network. Other requests are served in the meantime. When Postgres replies, the event loop resumes the original task.
Asynchronous drivers yield control during I/O, allowing massive concurrency.
SQLAlchemy 2.0 has built-in native support for asyncio. You use create_async_engine and AsyncSession, and you await database operations. The query syntax uses select(...) and session.execute(...) instead of the old legacy .query(...).
SQLAlchemy 2.0 handles async cleanly with select() and await session.execute().
SQLModel, created by FastAPI's author, combines SQLAlchemy (for the database) and Pydantic (for data validation) into a single model definition. You define a class once, and it serves as both the database table schema and the Pydantic data validator for your API endpoints.
SQLModel reduces boilerplate by uniting DB schemas and Pydantic validators.
Code examples
Defining an async model with SQLModel
from sqlmodel import SQLModel, Field
from typing import Optional
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = NoneYou define the model once. table=True makes it a SQLAlchemy model, while under the hood it remains a Pydantic model. We use modern Python typing.
Executing an async query
from sqlmodel import select
from sqlalchemy.ext.asyncio import AsyncSession
async def get_heroes(session: AsyncSession) -> list[Hero]:
# Built-in async support: we await the execute
statement = select(Hero).where(Hero.age > 25)
result = await session.execute(statement)
return result.scalars().all()Notice the `await session.execute(...)`. The asyncpg driver yields control back to the event loop while Postgres works. select() replaces the legacy session.query().
Key points
- Never use psycopg2 in an async app: It blocks the thread. Use asyncpg combined with SQLAlchemy's AsyncEngine.
- Await everything: Any call that touches the database (session.execute, session.commit) must be awaited.
- SQLModel = SQLAlchemy + Pydantic: Reduces boilerplate by keeping schema and validation in one single class definition.
Common mistakes
- Using a synchronous driver in an async framework: If you configure your DB URL with postgresql:// instead of postgresql+asyncpg://, SQLAlchemy will default to a synchronous driver, freezing the entire event loop on every query.
Recall questions
- Why is a synchronous database driver dangerous in FastAPI?
- Which driver should you use with Postgres in an async SQLAlchemy setup?
- What is the primary benefit of SQLModel?
Questions & answers
Our FastAPI application's latency spikes massively under load, even though CPU usage is low. We're using SQLAlchemy with psycopg2. What's the problem and how do we fix it?
psycopg2 is a synchronous driver. It blocks the event loop thread while waiting for I/O, queuing up all other requests. Switch to an asynchronous driver like asyncpg and use SQLAlchemy's AsyncEngine.
How does SQLAlchemy 2.0's query syntax differ from 1.4, specifically regarding async support?
SQLAlchemy 2.0 uses select() and session.execute(stmt) instead of session.query(). This separates the statement construction from the execution, allowing the execution to be safely awaited.