SQLAlchemy ORM & Transactions

Introduction

For production data layers, Python teams commonly use SQLAlchemy for modeling, querying, migrations, and transaction boundaries.

Understanding the topic

Core concepts to understand:

• Use declarative models with typed columns.
• Session = unit of work; keep transaction scope tight.
• Use explicit commit/rollback and retry transient errors.
• For async APIs, pair FastAPI with async SQLAlchemy engines.

Syntax reference

Visual flow / code:

from sqlalchemy import create_engine, Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

engine = create_engine("postgresql+psycopg://user:pass@localhost/app")

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)

Base.metadata.create_all(engine)

with Session(engine) as session:
    try:
        session.add(User(email="a@example.com"))
        session.commit()
    except Exception:
        session.rollback()
        raise

Execution workflow

Real-world use

Most Python microservices are API + Postgres. SQLAlchemy gives safe transaction handling and database portability without sacrificing SQL power.

Best practices

• One session per request/job.
• Use migrations (Alembic).
• Avoid long-lived transactions.
• Index frequently queried fields.

Common mistakes

• Global shared sessions cause race conditions.
• N+1 queries without eager loading.

Hands-on exercise

Interview preparation — practice these questions:

• Session lifecycle?
• What is N+1 and how to fix it?
• ORM vs raw SQL trade-offs?

Summary

In summary: Treat session as unit-of-work. Transaction boundaries are architecture decisions.