🗺️ ORM Basics with SQLAlchemy

What Is an ORM?

Object-Relational Mapping (ORM) bridges two worlds that think differently. Your Python code thinks in objects — classes, attributes, and references between them. Your database thinks in relations — tables, columns, rows, and foreign keys. An ORM translates automatically between the two, so you can write user.posts instead of a JOIN.

💡 An analogy — the universal translator: Your app speaks "Object-Oriented"; your database speaks "Relational". Without a translator you'd constantly convert objects to SQL and SQL results back to objects, by hand. The ORM is the universal translator sitting between them: you keep speaking your native language, and it handles every conversion behind the scenes.

SQLAlchemy's Two Layers

SQLAlchemy is really two libraries stacked together, and you can use either or both.

• Core — a SQL toolkit: the Engine, connection pooling, and a Pythonic SQL expression language. You can use Core alone for a lightweight, explicit approach.
• ORM — built on Core: maps classes to tables, tracks changes in a Session, and manages relationships.

This lesson focuses on the ORM, but note that the modern select() construct you'll use comes straight from Core — the two layers now share one unified querying style.

The Engine & Connection Setup

Install SQLAlchemy alongside the PostgreSQL driver:

python -m venv venv
source venv/bin/activate            # Windows: venv\Scripts\activate
pip install "sqlalchemy[postgresql]" psycopg

The Engine is the entry point to the database. Create one per application — it owns the connection pool. Read credentials from the environment rather than hardcoding them:

import os
from sqlalchemy import create_engine

# Dialect+driver://user:password@host:port/database
url = os.environ.get(
    "DATABASE_URL",
    "postgresql+psycopg://appuser:dev_password@localhost:5432/blog",
)

engine = create_engine(
    url,
    pool_size=5,        # connections kept open
    max_overflow=10,    # extra connections allowed under load
    pool_pre_ping=True, # check a connection is alive before using it
    echo=False,         # set True in development to see the SQL it emits
)

Defining Models

In SQLAlchemy 2.0, models subclass a DeclarativeBase and declare columns with typed Mapped[...] annotations plus mapped_column(...). The type hints double as real Python types and the column definition — one source of truth.

from datetime import datetime, timezone
from sqlalchemy import String, ForeignKey, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    """All models inherit from this."""


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    password_hash: Mapped[str] = mapped_column(String(128))
    is_active: Mapped[bool] = mapped_column(default=True)
    # server_default=func.now() lets the database fill the timestamp
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    # One user has many posts (the relationship, defined once)
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

    def __repr__(self) -> str:
        return f"<User {self.username!r}>"


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column()          # maps to TEXT
    is_published: Mapped[bool] = mapped_column(default=False)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    # Each post belongs to one user
    author: Mapped["User"] = relationship(back_populates="posts")

    def __repr__(self) -> str:
        return f"<Post {self.title!r}>"


# Create the tables (reads every model that inherits from Base)
Base.metadata.create_all(engine)

Common column types

The Session & CRUD

The Session is your unit of work — a staging area that tracks new, changed, and deleted objects, then flushes them to the database when you commit. Create a session factory once, bound to the engine:

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

Then open a session per unit of work with a with block. Wrapping it in Session.begin() gives you automatic commit-on-success and rollback-on-error:

Create

# begin() commits at the end of the block, or rolls back on exception
with Session.begin() as session:
    user = User(
        username="johndoe",
        email="john@example.com",
        password_hash="not_a_real_hash",
    )
    session.add(user)
    # After flush/commit, the database-generated id is populated
# print works after commit because the object was refreshed
print(user.id)

Read — the modern select() API

SQLAlchemy 2.0 unifies querying around select(). Build a statement, then execute it with session.scalars() (which returns model objects) or session.execute() (which returns rows):

from sqlalchemy import select

with Session() as session:
    # Fetch by primary key — the quickest lookup
    user = session.get(User, 1)

    # One row (or None)
    stmt = select(User).where(User.username == "johndoe")
    user = session.scalars(stmt).first()

    # Many rows with filtering and ordering
    stmt = (
        select(User)
        .where(User.is_active.is_(True))
        .where(User.email.like("%@example.com"))
        .order_by(User.created_at.desc())
        .limit(10)
    )
    users = session.scalars(stmt).all()
    for u in users:
        print(u.username, u.email)

Update

Because the Session tracks loaded objects, updating is often just assigning an attribute — SQLAlchemy notices the change and writes it on commit:

with Session.begin() as session:
    user = session.scalars(
        select(User).where(User.username == "johndoe")
    ).first()
    if user:
        user.email = "john.doe@example.com"   # tracked automatically
    # committed on block exit

Delete

with Session.begin() as session:
    user = session.get(User, 1)
    if user:
        session.delete(user)

Relationships

Relationships are the ORM's superpower: navigate between related objects in Python without writing JOINs. You already saw one-to-many above (a user has many posts). Here's the full picture.

Many-to-many with an association table

from sqlalchemy import Table, Column, ForeignKey

# The join table linking posts and tags
post_tag = Table(
    "post_tag",
    Base.metadata,
    Column("post_id", ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)


class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)

    posts: Mapped[list["Post"]] = relationship(
        secondary=post_tag, back_populates="tags"
    )


# Add to Post:
#     tags: Mapped[list["Tag"]] = relationship(
#         secondary=post_tag, back_populates="posts"
#     )

Using relationships

with Session.begin() as session:
    author = User(username="blogger", email="blog@example.com", password_hash="x")
    # Attach posts straight to the collection — cascade saves them
    author.posts = [
        Post(title="First Post", content="Hello, world."),
        Post(title="Second Post", content="More thoughts."),
    ]
    session.add(author)   # posts are saved along with the user

Avoiding the N+1 problem with eager loading

Looping over users and touching user.posts fires one extra query per user. Load everything up front with selectinload:

from sqlalchemy.orm import selectinload

with Session() as session:
    stmt = select(User).options(selectinload(User.posts))
    for user in session.scalars(stmt):
        # No extra query here — posts were already loaded
        print(f"{user.username}: {len(user.posts)} posts")

PostgreSQL-Specific Columns

The ORM doesn't hide PostgreSQL's best features. Import types from the postgresql dialect to use JSONB and arrays as first-class mapped columns.

from sqlalchemy.dialects.postgresql import JSONB, ARRAY
from sqlalchemy import String


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    tags: Mapped[list[str]] = mapped_column(ARRAY(String))
    attributes: Mapped[dict] = mapped_column(JSONB, default=dict)

Insert Python lists and dicts directly — SQLAlchemy adapts them:

with Session.begin() as session:
    session.add(Product(
        name="Ergonomic Keyboard",
        tags=["electronics", "office", "ergonomic"],
        attributes={"layout": "split", "wireless": True},
    ))

And query with PostgreSQL's operators through SQLAlchemy:

with Session() as session:
    # Array: products carrying a specific tag (uses PostgreSQL's ANY)
    stmt = select(Product).where(Product.tags.any("office"))
    office = session.scalars(stmt).all()

    # JSONB containment: attributes contain {"wireless": true}  (the @> operator)
    stmt = select(Product).where(Product.attributes.contains({"wireless": True}))
    wireless = session.scalars(stmt).all()

    # JSONB field access: attributes ->> 'layout' == 'split'
    stmt = select(Product).where(Product.attributes["layout"].astext == "split")
    split = session.scalars(stmt).all()

Hands-on Exercise

from datetime import datetime
from sqlalchemy import create_engine, String, ForeignKey, select, func
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship,
    sessionmaker, selectinload,
)

engine = create_engine("postgresql+psycopg://appuser:dev_password@localhost/blog")
Session = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    posts: Mapped[list["Post"]] = relationship(back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column()
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    author: Mapped["User"] = relationship(back_populates="posts")


Base.metadata.create_all(engine)

# Step 2: create a user with two posts
with Session.begin() as session:
    ana = User(username="ana", email="ana@example.com")
    ana.posts = [
        Post(title="Hello", content="First!"),
        Post(title="Again", content="Second!"),
    ]
    session.add(ana)

# Steps 3 & 4: newest-first posts by ana, with author preloaded
with Session() as session:
    stmt = (
        select(Post)
        .join(Post.author)
        .where(User.username == "ana")
        .order_by(Post.created_at.desc())
        .options(selectinload(Post.author))
    )
    for post in session.scalars(stmt):
        print(f"{post.title} — by {post.author.username}")   # no N+1

Summary & Quiz

📚 Further Reading

• SQLAlchemy 2.0 ORM Quick Start
• SQLAlchemy Unified Tutorial
• SQLAlchemy — PostgreSQL Dialect

🚀 What's Next?

You've now covered the relational world end to end — the database, the driver, and the ORM. Next, MongoDB Document Structure steps into NoSQL, where data lives as flexible documents instead of rows and columns.