sqlalchemy β SQL Toolkit & ORM#
What it is#
SQLAlchemy provides two layers:
- Core β SQL expression language that constructs queries as Python objects.
- ORM β maps Python classes to tables;
Sessionis the unit of work.
SQLAlchemy 2.0 unified the API and requires explicit with Session(engine) as s: patterns. It supports SQLite, PostgreSQL, MySQL, Oracle, and more.
Install#
pip install sqlalchemy
# Database-specific drivers (install alongside SQLAlchemy)
pip install psycopg2-binary # PostgreSQL
pip install pymysql # MySQL
# SQLite is built into Python β no extra driver needed
Quick example β Core#
from sqlalchemy import create_engine, text
engine = create_engine("sqlite+pysqlite:///:memory:", echo=False)
with engine.connect() as conn:
conn.execute(text("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)"))
conn.execute(text("INSERT INTO users VALUES (1, 'Alice'), (2, 'Bob')"))
conn.commit()
rows = conn.execute(text("SELECT * FROM users")).fetchall()
for row in rows:
print(row)
Output:
(1, 'Alice')
(2, 'Bob')
When / why to use it#
- You want a Pythonic, database-agnostic query interface with good migration tooling (Alembic).
- Building a web app with dynamic queries and relationship traversal.
- When you need both Core (low-level control) and ORM (convenience) in the same project.
Common pitfalls#
[!WARNING] N+1 query problem β lazy-loading relationships fires one SQL query per ORM object when you iterate. Fix with eager loading:
select(User).options(selectinload(User.posts)).
[!WARNING]
Sessionis not thread-safe β one session per request (web) or per thread. Never share aSessionacross threads.
[!WARNING]
text()wrapper required β SQLAlchemy 2.0 requirestext()around raw SQL strings. Bare strings will raiseObjectNotExecutableError.
Richer example β ORM with 2.0-style declarations#
from sqlalchemy import create_engine, String, ForeignKey, select
from sqlalchemy.orm import (
DeclarativeBase, Mapped, mapped_column, relationship, Session
)
class Base(DeclarativeBase):
pass
class Department(Base):
__tablename__ = "department"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
employees: Mapped[list["Employee"]] = relationship(back_populates="department")
class Employee(Base):
__tablename__ = "employee"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
dept_id: Mapped[int] = mapped_column(ForeignKey("department.id"))
department: Mapped["Department"] = relationship(back_populates="employees")
engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)
with Session(engine) as session:
eng = Department(id=1, name="Engineering")
session.add_all([
eng,
Employee(id=1, name="Alice", dept_id=1),
Employee(id=2, name="Bob", dept_id=1),
])
session.commit()
stmt = select(Employee).where(Employee.name.startswith("A"))
for emp in session.scalars(stmt):
print(f"{emp.name} β {emp.department.name}")
Output:
Alice β Engineering
Connection strings#
# SQLite (file)
engine = create_engine("sqlite:///app.db")
# SQLite (in-memory)
engine = create_engine("sqlite+pysqlite:///:memory:")
# PostgreSQL
engine = create_engine("postgresql+psycopg2://user:pass@localhost/dbname")
# MySQL
engine = create_engine("mysql+pymysql://user:pass@localhost/dbname")
Quick reference#
| Task | Code |
|---|---|
| Create engine | create_engine(url, echo=True) |
| Run raw SQL | conn.execute(text("SELECT 1")) |
| Insert ORM | session.add(obj) / session.add_all([...]) |
| Commit | session.commit() |
| Query all | session.scalars(select(Model)).all() |
| Filter | .where(Model.col == val) |
| Update | session.execute(update(Model).where(...).values(...)) |
| Delete | session.execute(delete(Model).where(...)) |
| Get by PK | session.get(Model, pk_value) |