skip to content

sqlalchemy β€” SQL Toolkit & ORM

Connect to databases, write queries, and define ORM models with SQLAlchemy 2.0. Covers the engine, sessions, Core queries, ORM declarative models, and the N+1 pitfall.

3 min read 6 snippets yesterday intermediate

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; Session is 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] Session is not thread-safe β€” one session per request (web) or per thread. Never share a Session across threads.

[!WARNING] text() wrapper required β€” SQLAlchemy 2.0 requires text() around raw SQL strings. Bare strings will raise ObjectNotExecutableError.

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#

TaskCode
Create enginecreate_engine(url, echo=True)
Run raw SQLconn.execute(text("SELECT 1"))
Insert ORMsession.add(obj) / session.add_all([...])
Commitsession.commit()
Query allsession.scalars(select(Model)).all()
Filter.where(Model.col == val)
Updatesession.execute(update(Model).where(...).values(...))
Deletesession.execute(delete(Model).where(...))
Get by PKsession.get(Model, pk_value)