FastAPI + SQLAlchemy 2.0: Transaction design to avoid dying in the asynchronous “democracy”
It's 2 a.m. and my heart is pounding from the caffeine from my fourth cup of coffee. During this time, when I was relying only on the light from the monitor, I was at the mercy of FastAPI and SQLAlchemy 2.0's asynchronous processing, and it took me another three hours.
"Using async/await will make it faster." I want to punch myself for being lured in by such sweet words and starting implementation. Behind the sophisticated syntax of SQLAlchemy 2.0 lies the "asynchronous trap" unique to ORMs. If you write it properly, a hellish situation called deadlock awaits you in the production environment.
Today, I'm going to share the "survival strategy for asynchronous DB operations" that I've learned through many late-night debugging sessions over the past few years.
1. Session management that kills “it doesn’t work for some reason”
First is the classic DI (dependency injection) pattern. Are you not satisfied with copying the official documents? The important question is why this setting is necessary.
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
# echo=True is required during development. There is no future for engineers who cannot visually check the order in which SQL is generated.
engine = create_async_engine(DATABASE_URL, echo=True, pool_size=10)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
async def get_db():
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
exception Exception:
await session.rollback()
raise
finally:
# If you don't explicitly close it here, the connection pool will be exhausted and you will die
await session.close()
Why `expire_on_commit=False`?
If this is not set, all objects in the session will become "expired" the moment you commit. If you try to access the attribute immediately after that, you'll be hit with an error called `DetachedInstanceError` that says, ``What are you trying to do after your session expires?'' If you watch this late at night, you'll be really disappointed.2. The real cause of deadlock is "waiting for IO"
Last week, I got really into the balance update process. The question is "Why does it deadlock when using `select().with_for_update()`?" The simple answer is that you were inserting unnecessary IO within the scope of the transaction. **
async def update_balance(session: AsyncSession, user_id: int, amount: int):
# get row lock
stmt = select(User).filter_by(id=user_id).with_for_update()
result = await session.execute(stmt)
user = result.scalar_one()
# [Danger] If you hit an external API or perform heavy calculations here, the lock will not be released.
# Another thread/coroutine will wait on this line and the deadlock will be completed
user.balance += amount
await session.commit()
Lesson learned: Never perform "external communication" while the DB is locked. The golden rule for ORM transactions is to keep them as short as possible.
3. Preventing “accidents” with architecture: Introducing the Repository pattern
``Directly writing DB logic in the controller (FastAPI path manipulation function)'' will collapse the moment the code base becomes large. I isolate DB operations to a dedicated Repository layer and forcefully control transaction granularity.class UserRepository:
def __init__(self, session: AsyncSession):
self.session = session
async def get_user_for_update(self, user_id: int):
stmt = select(User).where(User.id == user_id).with_for_update()
res = await self.session.execute(stmt)
return res.scalar_one()
# Manage transactions in the Service layer
async def transfer_funds(db: AsyncSession, from_id: int, to_id: int, amount: int):
repo = UserRepository(db)
# Strictly control the order here (avoid deadlock by taking locks in ascending order of ID)
# ...
4. Async “don’t” list
I will summarize my blood lessons for mastering SQLAlchemy 2.0.
| Prohibitions | Reasons |
|---|---|
| Reusing the same session with `asyncio.gather` | `AsyncSession` is not coroutine safe. Internal states are corrupted due to conflicts. |
| `Use session.query()` | Legacy of 1.x series. For 2.0 series, `select()` is the only option. |
| Get a tuple with `result.all()` | If you forget `scalars()`, a Python tuple will be returned and an attribute error will occur. |
| Use `scoped_session` asynchronously | `scoped_session` is designed to be thread-local. In an asynchronous environment, if you are not aware of `task_local` (asyncio context), it will leak. |
Checklist that can be used starting tomorrow
1. Did you check the SQL? : Is the SQL issued with `echo=True` committed at the expected timing?
2. Is the lock range minimum? : Is there any external communication with `await` after `with_for_update()`?
3. Are there any reuses of sessions? : Are you using the same session in `asyncio.gather`?
4. Have you forgotten `scalars()`? : Is it possible to obtain it as an object instead of `result.all()`?
After all, ORM isn't magic. It's training to be able to simulate the SQL generated by SQLAlchemy in your head.
Late night coffee has almost run out. The quality of code is not ``how easy it is to make it work,'' but ``how to hide the dirty parts and design a safe API.'' Tomorrow is also early, so I'll commit here today and go to sleep.
SQLAlchemy 2.0 is, after all, a tool for accurately communicating the intent of SQL. If you can do that, it will become the most powerful weapon.