Posted inORM Python

SQL Alchemy

What is SQLAlchemy?

SQLAlchemy is Python’s most popular database toolkit. It has two big layers:

  • SQLAlchemy Core: build SQL expressions and run them (closer to SQL).
  • SQLAlchemy ORM: map Python classes to database tables and work with objects.

In SQLAlchemy 2.x, the “2.0 style” API is the recommended way: select(...), Session as a context manager, typed mappings, etc.

1) Install + pick a database

Install

pip install sqlalchemy

If you use Postgres:

pip install psycopg[binary]

If you use MySQL:

pip install pymysql

For this tutorial we’ll use SQLite because it needs no server.

2) Create an engine (the DB connection factory)

from sqlalchemy import create_engine

engine = create_engine("sqlite:///app.db", echo=True)
  • sqlite:///app.db creates a file DB named app.db.
  • echo=True prints SQL to the console (great for learning).

3) Define ORM models (tables as Python classes)

SQLAlchemy 2.x uses typed ORM mappings via Mapped[...] and mapped_column(...).

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, ForeignKey
from typing import List, Optional

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String(100), nullable=False)

    # one-to-many: User -> Post
    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), nullable=False)
    body: Mapped[Optional[str]]

    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
    author: Mapped["User"] = relationship(back_populates="posts")

Key points:

  • __tablename__ sets the table name.
  • mapped_column(primary_key=True) creates an integer PK by default.
  • ForeignKey("users.id") enforces relational integrity.
  • relationship(...) links objects in Python.

4) Create tables

Base.metadata.create_all(engine)

Run that once (or whenever you change models in this simple tutorial). In real projects you’ll use Alembic migrations (covered later).

5) Sessions: the unit of work

A Session is how you talk to the ORM: persist objects, run queries, manage transactions.

from sqlalchemy.orm import Session

with Session(engine) as session:
    # work here
    session.commit()

Common pattern: “open session → do work → commit”.

6) Insert data

from sqlalchemy.orm import Session

with Session(engine) as session:
    u1 = User(email="romaan@example.com", name="Romaan")
    u2 = User(email="alex@example.com", name="Alex")

    session.add_all([u1, u2])
    session.commit()

Insert related objects

with Session(engine) as session:
    user = session.query(User).filter_by(email="romaan@example.com").one()

    post = Post(title="Hello SQLAlchemy", body="First post", author=user)
    session.add(post)
    session.commit()

7) Querying with select() (recommended in 2.x)

from sqlalchemy import select
from sqlalchemy.orm import Session

with Session(engine) as session:
    stmt = select(User).where(User.email == "romaan@example.com")
    user = session.execute(stmt).scalar_one()
    print(user.id, user.name)

Get many rows

with Session(engine) as session:
    stmt = select(User).order_by(User.id)
    users = session.execute(stmt).scalars().all()
    for u in users:
        print(u.email)

Query posts with join

from sqlalchemy import select

with Session(engine) as session:
    stmt = (
        select(Post, User)
        .join(Post.author)
        .where(User.email == "romaan@example.com")
    )
    rows = session.execute(stmt).all()
    for post, user in rows:
        print(user.name, "->", post.title)

8) Update data

Update by loading object then editing

from sqlalchemy import select
from sqlalchemy.orm import Session

with Session(engine) as session:
    user = session.execute(select(User).where(User.email == "alex@example.com")).scalar_one()
    user.name = "Alexandra"
    session.commit()

Bulk update (Core-style)

from sqlalchemy import update
from sqlalchemy.orm import Session

with Session(engine) as session:
    session.execute(
        update(User)
        .where(User.email == "alex@example.com")
        .values(name="Alexandra")
    )
    session.commit()

Bulk updates don’t automatically update already-loaded objects in the session unless you refresh them.

9) Delete data

from sqlalchemy import select
from sqlalchemy.orm import Session

with Session(engine) as session:
    post = session.execute(select(Post).where(Post.title == "Hello SQLAlchemy")).scalar_one()
    session.delete(post)
    session.commit()

10) Relationships and loading (the N+1 problem)

If you fetch users and then access user.posts for each user, you might accidentally run many queries.

Eager load with selectinload

from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload

with Session(engine) as session:
    stmt = select(User).options(selectinload(User.posts))
    users = session.execute(stmt).scalars().all()

    for u in users:
        print(u.email, "posts:", [p.title for p in u.posts])
  • selectinload usually runs 2 queries total: one for users, one for posts.

11) Transactions + rollback

SQLAlchemy sessions are transactional.

from sqlalchemy.orm import Session

try:
    with Session(engine) as session:
        session.add(User(email="romaan@example.com", name="Duplicate"))  # violates unique
        session.commit()
except Exception as e:
    print("Failed:", e)

You can also do explicit transaction blocks:

from sqlalchemy.orm import Session

with Session(engine) as session:
    with session.begin():  # auto-commit or rollback
        session.add(User(email="new@example.com", name="New"))

12) Schema migrations with Alembic (real-world essential)

Install:

pip install alembic

Initialize in your project:

alembic init alembic

Set DB URL in alembic.ini:

sqlalchemy.url = sqlite:///app.db

Configure alembic/env.py to import your Base and set target_metadata = Base.metadata.

Create a migration:

alembic revision --autogenerate -m "create users and posts"

Apply it:

alembic upgrade head

(If you want, tell me your project layout and I’ll show you the exact env.py wiring.)

13) Patterns you’ll use constantly

A) “Repository-ish” helper: get by id

from sqlalchemy import select
from sqlalchemy.orm import Session

def get_user(session: Session, user_id: int) -> User | None:
    return session.execute(select(User).where(User.id == user_id)).scalar_one_or_none()

B) Pagination

from sqlalchemy import select

def list_users(session: Session, offset: int = 0, limit: int = 20):
    stmt = select(User).order_by(User.id).offset(offset).limit(limit)
    return session.execute(stmt).scalars().all()

C) Counting

from sqlalchemy import select, func

with Session(engine) as session:
    count = session.execute(select(func.count(User.id))).scalar_one()
    print(count)

14) Async SQLAlchemy (quick intro)

If you’re using FastAPI and async endpoints, you’ll likely use AsyncEngine + AsyncSession.

Install an async driver:

  • SQLite async: pip install aiosqlite
  • Postgres async: pip install asyncpg

Example:

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

engine = create_async_engine("sqlite+aiosqlite:///app.db", echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def main():
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User))
        users = result.scalars().all()
        print(users)

asyncio.run(main())

15) Common mistakes (and how to avoid them)

  • Forgetting to commit → inserts/updates don’t persist.
  • Using ORM objects outside a session (or after session closes) → lazy loads fail.
    • Fix: eager load (selectinload) or keep session scope correct.
  • N+1 queries → load relationships with selectinload / joinedload.
  • Not using migrations → production pain. Use Alembic.

A complete runnable mini example

Put this in main.py and run it.

from typing import List, Optional
from sqlalchemy import create_engine, String, ForeignKey, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session, selectinload

engine = create_engine("sqlite:///app.db", echo=True)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    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), nullable=False)
    body: Mapped[Optional[str]]
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
    author: Mapped["User"] = relationship(back_populates="posts")

def seed():
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        # Only add if empty
        existing = session.execute(select(User)).scalars().first()
        if existing:
            return

        u1 = User(email="romaan@example.com", name="Romaan")
        u2 = User(email="alex@example.com", name="Alex")
        session.add_all([u1, u2])
        session.flush()  # get ids without commit

        session.add_all([
            Post(title="Hello SQLAlchemy", body="First post", author=u1),
            Post(title="Second post", body=None, author=u1),
        ])
        session.commit()

def demo_queries():
    with Session(engine) as session:
        users = session.execute(
            select(User).options(selectinload(User.posts)).order_by(User.id)
        ).scalars().all()

        for u in users:
            print(u.email, "=>", [p.title for p in u.posts])

if __name__ == "__main__":
    seed()
    demo_queries()

If you tell me which stack you’re using (SQLite vs Postgres, sync vs async, FastAPI vs Flask vs plain scripts), I’ll tailor this into a “real project” layout (config file, session factory, dependency injection pattern, and Alembic setup).