9. Databases & Transactions (SQLAlchemy)

Design with correctness and throughput in mind: manage sessions explicitly, choose isolation levels, and avoid N+1.

Question: How does SQLAlchemy 2.0 style differ from older versions, and why is it preferred?

Answer: SQLAlchemy 2.0 style promotes a more explicit, "Core-first" approach even when using the ORM. Queries are constructed using select() statements rather than session.query(), and results are accessed through a Result object. This creates a clearer, more consistent API between the Core and ORM layers.

Explanation: The 2.0 style is fully type-hinted, making it much easier to work with static analysis tools like mypy. It also encourages a clearer separation of concerns, with session management being more explicit. For asynchronous operations with asyncio, it uses drivers like asyncpg.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

# Assume User model exists
async def get_user(session: AsyncSession, user_id: int):
    stmt = select(User).where(User.id == user_id)
    result = await session.execute(stmt)
    return result.scalar_one()

Question: What are database transaction isolation levels?

Answer: Isolation levels define the degree to which one transaction must be isolated from the data modifications made by any other transaction. Common levels, from weakest to strongest, are READ UNCOMMITTED, READ COMMITTED, REPEATABLE READ, and SERIALIZABLE. The default for databases like PostgreSQL is READ COMMITTED. Senior engineers should understand the trade-offs between performance and consistency that each level provides.

Question: How do you avoid the N+1 problem in ORMs?

Answer: Use eager loading strategies like selectinload/joinedload to prefetch related rows.

Explanation: This reduces the number of round trips and improves performance.

from sqlalchemy.orm import selectinload
stmt = select(User).options(selectinload(User.posts))

Question: What is the recommended session pattern in SQLAlchemy 2.x (async)?

Answer: Use a single AsyncEngine and short-lived AsyncSessions via async with to scope transactions.

Explanation: Ensures connections return to the pool and transactions close deterministically.

from sqlalchemy.ext.asyncio import async_sessionmaker
Session = async_sessionmaker(engine, expire_on_commit=False)
async with Session.begin() as s:
    s.add(obj)

Question: How do you implement optimistic concurrency control?

Answer: Add a version column and check it on update to detect lost updates.

Explanation: SQLAlchemy supports version_id_col to automate checks.

from sqlalchemy.orm import DeclarativeBase, mapped_column
class Base(DeclarativeBase): pass
class Order(Base):
    __tablename__ = "orders"
    id = mapped_column(primary_key=True)
    version = mapped_column(default=0)
# compare-and-swap in UPDATE WHERE id=? AND version=?

Question: How do you handle serialization failures and deadlocks robustly?

Answer: Detect retriable errors (e.g., PostgreSQL 40001/40P01) and retry the transaction with backoff and jitter; keep transactions short to reduce contention.

import random, time
for attempt in range(5):
    try:
        with Session.begin() as s:
            do_work(s)
        break
    except DBError as e:
        if e.is_retriable():
            time.sleep(0.05 * (2 ** attempt) + random.random()/100)
            continue
        raise