SQLAlchemy 2.0 basics: a practical Python Backend guide

SQLAlchemy 2.0 basics: a practical Python Backend guide

After years of production applications built on SQLAlchemy 1.4, the 2.0 release brings a modern, fully-typed API that makes database interactions safer and more predictable. We've migrated several e-commerce and SaaS backends to 2.0, and the jump in code clarity alone justifies the effort. This guide walks you through the essentials: declarative models with type hints, the new select() API, session management, and async support—everything you need to build reliable data layers.

DeclarativeBase with Mapped types

The foundation of SQLAlchemy 2.0 is DeclarativeBase, which replaces the older declarative_base() factory. The key improvement is native support for Python type hints through the Mapped generic type. This gives you IDE autocomplete, static type checking with mypy, and runtime clarity about your schema.

Let's define a simple User model as we'd see in a real SaaS application:

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

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True)
    username: Mapped[str]
    is_active: Mapped[bool] = mapped_column(default=True)
    bio: Mapped[Optional[str]] = None

Notice the Mapped[T] wrapper around each column type. This tells both mypy and SQLAlchemy what the Python type of each column is. The mapped_column() function replaces the older Column() syntax and is fully compatible with type hints.

When you use Mapped[Optional[str]], SQLAlchemy automatically makes that column nullable in the database. If you use Mapped[str], it's non-nullable. This removes a whole class of bugs where the Python type and database constraint were out of sync.

For a more complex scenario, imagine a ride-hailing backend with relationships:

from datetime import datetime
from sqlalchemy.orm import relationship

class Driver(Base):
    __tablename__ = "drivers"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    phone: Mapped[str] = mapped_column(unique=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    
    # Relationship
    trips: Mapped[list["Trip"]] = relationship(back_populates="driver")

class Trip(Base):
    __tablename__ = "trips"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    driver_id: Mapped[int] = mapped_column(ForeignKey("drivers.id"))
    pickup_location: Mapped[str]
    dropoff_location: Mapped[str]
    
    # Relationship
    driver: Mapped[Driver] = relationship(back_populates="trips")

The Mapped[list["Trip"]] syntax tells mypy exactly what type you get when you access driver.trips. Your editor can autocomplete attributes on Trip objects, and type checkers catch misuse before runtime.

select() for queries

SQLAlchemy 2.0's select() builder is a composable, fully-typed way to construct queries. It replaces the older query() interface and integrates seamlessly with modern async code.

Basic selection looks clean and familiar:

from sqlalchemy import select

# Get all users
stmt = select(User)
users = session.scalars(stmt).all()

# Get a single user by email
stmt = select(User).where(User.email == "alice@example.com")
user = session.scalar(stmt)

# Filter by multiple conditions
stmt = select(User).where(
    (User.is_active == True) &
    (User.email.like("%@company.com"))
)
active_company_users = session.scalars(stmt).all()

Notice that User.email gives you a typed column object with methods like .like(), and the where() clause uses standard boolean operators (&, |, ~). Your IDE knows what's available, preventing typos.

For joins, the syntax is explicit and readable:

stmt = select(Trip, Driver).join(Driver).where(
    Trip.pickup_location == "Downtown"
)
results = session.execute(stmt).all()

# results is a list of tuples: [(trip, driver), ...]
for trip, driver in results:
    print(f"{driver.phone} picked up from {trip.pickup_location}")

Ordering, limiting, and aggregations follow the same pattern:

from sqlalchemy import func

# Top 10 most active drivers
stmt = select(Driver, func.count(Trip.id).label("trip_count")).join(Trip).group_by(Driver.id).order_by(func.count(Trip.id).desc()).limit(10)

# Count users
stmt = select(func.count()).select_from(User)
total = session.scalar(stmt)

The select() API is composable, so you can build queries incrementally—useful when filtering logic spans multiple functions or depends on user input.

Session per request

SQLAlchemy sessions manage your database connections and the identity map (keeping a single instance for each database row). The standard pattern is one session per HTTP request, which keeps state isolated and prevents accidental mutations across requests.

In a FastAPI application (common in our SaaS backends), you'd use dependency injection:

from sqlalchemy.orm import Session
from fastapi import FastAPI, Depends
from contextlib import contextmanager

app = FastAPI()
engine = create_engine("postgresql://user:pass@localhost/dbname")

def get_session():
    session = Session(engine)
    try:
        yield session
    finally:
        session.close()

@app.post("/users")
def create_user(email: str, session: Session = Depends(get_session)):
    user = User(email=email)
    session.add(user)
    session.commit()
    return user

@app.get("/users/{user_id}")
def get_user(user_id: int, session: Session = Depends(get_session)):
    return session.get(User, user_id)

The session is created per request, used for all queries, then closed. FastAPI cleans up automatically when the request finishes. This isolates transactions—one slow request doesn't block others.

If you're building a CLI tool or background job, you'd use explicit session management:

from sqlalchemy.orm import Session

def process_pending_trips():
    with Session(engine) as session:
        stmt = select(Trip).where(Trip.status == "pending")
        pending = session.scalars(stmt).all()
        
        for trip in pending:
            trip.status = "processing"
        
        session.commit()

The context manager pattern automatically commits on success and rolls back on exception. Simple and safe.

Async support with AsyncSession

For high-concurrency applications (like our ride-hailing platform handling thousands of simultaneous requests), AsyncSession lets you await database operations instead of blocking threads.

First, create an async engine:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# Use an async driver: psycopg (PostgreSQL) or asyncmy (MySQL)
async_engine = create_async_engine(
    "postgresql+psycopg://user:pass@localhost/dbname",
    pool_size=20,
    echo=False
)

async_session_maker = sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

Then use it in your async handlers:

from fastapi import FastAPI

app = FastAPI()

async def get_async_session():
    async with async_session_maker() as session:
        yield session

@app.get("/drivers/{driver_id}/trips")
async def get_driver_trips(driver_id: int, session: AsyncSession = Depends(get_async_session)):
    stmt = select(Trip).where(Trip.driver_id == driver_id)
    result = await session.execute(stmt)
    return result.scalars().all()

@app.post("/trips")
async def create_trip(
    driver_id: int, 
    pickup: str, 
    dropoff: str,
    session: AsyncSession = Depends(get_async_session)
):
    trip = Trip(driver_id=driver_id, pickup_location=pickup, dropoff_location=dropoff)
    session.add(trip)
    await session.commit()
    return trip

Key difference: you await the execute() and commit() calls. The session doesn't block while waiting for the database—other requests are processed. For read-heavy APIs, this multiplies your throughput.

One important detail: async sessions default to expire_on_commit=False to avoid lazy-loading issues. Once you commit, attributes loaded in that transaction remain available without requiring another query.

Common mistakes

Forgetting to commit(). In SQLAlchemy, add() stages changes; they're not persisted until you call commit(). Many developers coming from ORMs that auto-commit are surprised by this.

Using mutable defaults incorrectly. If you write metadata: Mapped[dict] = {}, all new instances share the same dict object. Use default_factory:

from sqlalchemy import JSON

class Config(Base):
    __tablename__ = "configs"
    id: Mapped[int] = mapped_column(primary_key=True)
    settings: Mapped[dict] = mapped_column(JSON, default=dict)  # Calls dict() each time

Ignoring type hints in relationships. Forgetting to annotate relationship types means you lose IDE support and type checking. Always use Mapped[list["ChildModel"]] or Mapped[Optional[ChildModel]].

Lazy-loading in async code. If you access driver.trips after the session closes in async code, you'll get an error because the relationship hasn't been loaded. Use selectinload() to eagerly fetch:

from sqlalchemy.orm import selectinload

stmt = select(Driver).options(selectinload(Driver.trips))
drivers = await session.scalars(stmt)
# Now driver.trips is available even after the session closes

Key takeaways

• DeclarativeBase and Mapped[T] bring Python type hints into your schema, catching bugs at dev time.
• select() is the new query API—composable, explicit, and fully typed.
• One session per request isolates transaction state and prevents surprising mutations.
• AsyncSession unlocks high-concurrency applications without thread pools.
• Always commit explicitly; always use type hints on relationships; use selectinload to avoid async lazy-loading pitfalls.
• For production systems, add connection pooling config and consider using Alembic for migrations (outside this guide's scope).