SQLAlchemy and Alembic 2 cheatsheet

SQLAlchemy and Alembic 2: Updating and raw SQL

6 common SQLAlchemy and Alembic 2 tasks.

Update an object

Change attributes on a loaded object and commit; SQLAlchemy tracks the change.

user = session.get(User, 1)
user.email = "[email protected]"
session.commit()

Update many rows

Run one UPDATE statement for all matching rows.

from sqlalchemy import update

session.execute(update(User).where(User.email.is_(None)).values(email="unknown"))
session.commit()

Delete a row

Delete an object and commit.

session.delete(session.get(Post, 1))
session.commit()

Caution: Deletes are permanent after commit. Back up first if unsure.

Roll back on error

Undo pending changes when something fails.

try:
    session.add(User(name="Bad"))
    session.flush()
except Exception:
    session.rollback()
    raise

Run raw SQL safely

Use text() with bound parameters, never string formatting.

from sqlalchemy import text

rows = session.execute(text("SELECT * FROM users WHERE name = :name"), {"name": "Ada"}).all()

Use the async API

Create an async engine and session for asyncio apps such as FastAPI. Use an async driver like asyncpg.

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
Session = async_sessionmaker(engine, expire_on_commit=False)