FastAPI + SQLAlchemy 2.0: The illusion of async and “sloppy” optimization techniques to survive in the field
"Asynchronous is faster" - Have you ever believed in such a naive word and put into production a code that uses `async` repeatedly, resulting in the CPU usage rate skyrocketing and you breaking out in a cold sweat while responding to an incident late at night?
Let's be honest. Asynchronous implementation is not magic. In fact, if not handled properly, it can become a hotbed for bottlenecks, which is far worse than synchronous code. In particular, when using SQLAlchemy 2.0 with FastAPI, if you implement ORM without proper ORM etiquette, the connection pool will be exhausted, the event loop will be blocked, and the API will "stop" instead of "blazingly fast."
In this article, I will abandon textbook-like explanations and share the key points of implementation that I have learned from numerous failures and that will work in the field.
1. Why “async” can be a trap
The biggest enemy of asynchronous processing is ``blocking event loop''.
For example, if you make a mistake in ORM session management and synchronous I/O occurs in the background, all other requests will stop at that moment. This is the true nature of the "asynchronous but slow" phenomenon.
Golden rules for connection management
Beginners tend to use `scoped_session`, but using it in an async environment is suicide. The only correct answer is to use `async_sessionmaker`, which is fully synchronized with the request lifecycle.# database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
# Set pool_size appropriately and prepare for sudden spikes with max_overflow
engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/dbname",
pool_size=20,
max_overflow=10,
pool_pre_ping=True # Connection disconnection countermeasure
)
AsyncSessionLocal = async_sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
async def get_db():
async with AsyncSessionLocal() as session:
try:
yield session
finally:
await session.close()
`
[Important] Curse of `expire_on_commit=False`
If you forget this, the object will be detached after the commit and subsequent property accesses will run `Lazy Loading' and crash with a session error. This is the culprit behind the mass production of "code that somehow works." Set it explicitly.
2. Visualizing N+1 problems: The power of `selectinload`
Stopping thinking that ``ORM is convenient'' increases infrastructure costs. Please see the comparison below.
Inefficient patterns (N+1)
# To get 100 users and get each post in a loop
users = await db.execute(select(User))
for user in users.scalars():
posts = await user.awaitable_attrs.posts # SELECT is issued here every time
`
With this, if there are 100 users, 101 queries will be executed. The load on the DB becomes visibly worse.
Optimization patterns
from sqlalchemy.orm import selectinload
# Combine and get at once
stmt = select(User).options(selectinload(User.posts))
result = await db.execute(stmt)
`
This alone will drastically reduce the number of queries to 2. This may be an error for a small app, but when the number of users increases to several thousand, this difference is directly linked to server costs.
3. Breaking through the “wall” of 2.0 migration
The most frustrating thing about migrating from the 1.x series to the 2.0 series is the abolition of the `query()` method. Moving completely to the `select()` function may seem tedious at first.- 1.x: `session.query(User).filter(User.id == 1).first()`
- 2.0: `session.execute(select(User).where(User.id == 1)).scalar_one_or_none()`
4. Surviving the hell of asynchronous debugging
The stack trace of `asyncio` is almost like an unbreakable code. When the stack gets deep, you may have no idea where it stopped.
Recommended debugging techniques
1. Utilizing `pytest-asyncio`: Control the behavior of `loop` in the test code and write highly reproducible tests. 2. Structure the log: Assign a request ID using `structlog`, etc., so that asynchronous flows can be traced from the log. 3. `breakpoint()` instead of `pdb`: In Python 3.7 and later, use `breakpoint()` and get into the habit of checking what the event loop is doing on the spot.5. Finally: ORM is not a “silver bullet”
FastAPI + SQLAlchemy 2.0 is powerful, but the ORM abstraction becomes a hindrance in complex transaction management and extremely high-load environments.
- ORM is a tool to buy development efficiency, and if you want the ultimate performance, you need the courage to directly write SQL with `asyncpg`.
- Although migrating to 2.0 requires learning costs, think of it as an "investment in your future self" in terms of type safety and maintainability.
Well, readers. If you have any ``strange phenomena peculiar to asynchronous'' that occur in your environment or ``solutions to the quagmire of SQLAlchemy,'' please let us know in the comments. It is our knowledge from the field that makes us stronger.