CheatSQLAlchemy Python SQL toolkit
Light · Postcard
102 × 152 mm · PDF
Dark · Postcard
102 × 152 mm · PDF
Ink saver · Postcard
102 × 152 mm · PDF
Light · A4 landscape
297 × 210 mm · PDF
Dark · A4 landscape
297 × 210 mm · PDF
Ink saver · A4 landscape
297 × 210 mm · PDF
Light · A3 landscape
420 × 297 mm · PDF
Dark · A3 landscape
420 × 297 mm · PDF
Ink saver · A3 landscape
420 × 297 mm · PDF
Start here
create_engine("<url>")
Connect to a database
class User(Base)
Map a table to a class
Base.metadata.create_all(engine)
Create the tables
with Session(engine) as s:
Open a unit of work
s.add(obj)Queue an INSERT
s.scalars(select(User))
Query objects
s.commit()Flush and save
relationship()Link two classes
Concepts
- Engine URL + pool
- Connection Core, raw SQL
- Session ORM unit of work
- Mapped objects identity map
Engine
from sqlalchemy import create_engine
Import the engine factory
create_engine("sqlite:///app.db")
SQLite file engine
create_engine("postgresql+psycopg://<user>:<pw>@<host>/<db>")
PostgreSQL engine
Format is
dialect+driver://user:password@host:port/database.
create_engine("sqlite://", echo=True)
In-memory DB, log all SQL
from sqlalchemy import text
Wrap a raw SQL string
with engine.connect() as conn:
Core connection block
Autobegin: call conn.commit() to save; nothing is committed on
exit.
conn.execute(text("SELECT 1"))
Run raw SQL
conn.execute(text("... :id"), {"id":
5})
Bind parameters by name
Never format values into the string; use bound parameters.
with engine.begin() as conn:
Connection that commits on exit
Declare models
class Base(DeclarativeBase): pass
Declarative base class
__tablename__ = "user"
Table name for a model
id: Mapped[int] = mapped_column(primary_key=True)
Primary key column
name: Mapped[str] = mapped_column(String(50))
Typed string column
email: Mapped[Optional[str]]
Nullable column
Mapped[str] means NOT NULL; Optional makes it nullable.
mapped_column(unique=True, index=True)
Unique constraint / index
mapped_column(ForeignKey("user.id"))
Foreign key column
Base.metadata.create_all(engine)
Create missing tables
Does not alter existing tables; use Alembic for that.
Session
from sqlalchemy.orm import Session
Import Session
with Session(engine) as session:
Open a session
session.add(<obj>)
Add one object
session.add_all([<a>, <b>])
Add several objects
session.commit()
Write changes and end transaction
session.rollback()
Discard pending changes
session.delete(<obj>)
Mark an object for deletion
session.get(User, <pk>)
Fetch by primary key
with Session(engine) as s, s.begin():
Commit on clean exit
sessionmaker(engine)
Session factory for an app
Query
from sqlalchemy import select
Import select()
session.scalars(select(User)).all()
All rows as objects
select(User).where(User.name == "ann")
Filter rows
session.scalars(stmt).first()
First object or None
session.scalar(stmt)
First column of first row
session.scalars(stmt).one()
Exactly one row, else error
session.execute(select(User.id, User.name))
Rows of column tuples
select(User).order_by(User.name.desc())
Sort rows
select(User).limit(<n>).offset(<n>)
Paginate
.where(User.age >= 18, User.active.is_(True))
AND several conditions
.where(or_(<a>, <b>))
OR conditions
.where(User.id.in_([1, 2]))
Match a list
select(func.count()).select_from(User)
Count rows
select(Order.user_id,
func.sum(Order.total)).group_by(Order.user_id)
Aggregate per group
Insert, update, delete
insert(User).values(name="ann")
Core INSERT
session.execute(insert(User), [{"name":
"a"}, ...])
Bulk insert dicts
user.name = "bob"
Update via attribute, then commit
update(User).where(<cond>).values(name="x")
Bulk UPDATE
delete(User).where(<cond>)
Bulk DELETE
Relationships
posts: Mapped[list["Post"]] =
relationship(back_populates="author")
One-to-many side
author: Mapped["User"] =
relationship(back_populates="posts")
Many-to-one side
user_id: Mapped[int] =
mapped_column(ForeignKey("user.id"))
The foreign key behind it
relationship(cascade="all, delete-orphan")
Delete children with parent
relationship(secondary=<table>)
Many-to-many via table
user.posts.append(Post(title="x"))
Add child through collection
select(User).options(selectinload(User.posts))
Eager-load collection
Avoids the N+1 query pattern.
select(Post).options(joinedload(Post.author))
Eager-load by JOIN
Joins
select(User, Post).join(Post)
Inner join on the foreign key
select(User).join(User.posts)
Join along a relationship
select(User).join(Post, User.id == Post.user_id)
Join with explicit condition
select(User).join(User.posts, isouter=True)
LEFT OUTER JOIN
Alembic migrations
pip install alembic
Install Alembic
alembic init <dir>
Create migration environment
target_metadata = Base.metadata
Set in env.py for autogenerate
alembic revision --autogenerate -m
"<msg>"
Diff models against DB
Review the generated file: renames show up as drop + add.
alembic upgrade head
Apply all migrations
alembic downgrade -1
Undo the last migration
alembic currentShow DB revision
Recipes
class Base(DeclarativeBase): passclass User(Base): __tablename__ = "user" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str]engine = create_engine("sqlite:///app.db")Base.metadata.create_all(engine)with Session(engine) as s: s.add(User(name="ann")) s.commit() for u in s.scalars(select(User)): print(u.id, u.name)stmt = (select(User.name, func.count(Post.id)) .join(User.posts, isouter=True) .group_by(User.id))rows = session.execute(stmt).all()SessionLocal = sessionmaker(engine)def get_db(): with SessionLocal() as s: yield s
