PythonMastery
intermediate 22 min read · lesson 3 of 12 in Python How-To

Databases: SQLite and SQLAlchemy

1 · The lesson

read

Every non-trivial app eventually needs a database — something that survives restarts, supports concurrent reads, and lets you ask "which users signed up last week?" without scanning a 4-GB JSON file. Python ships with SQLite in the standard library, which means you can write your first real database app today without installing anything.

This lesson covers the stdlib sqlite3 module first — parameterised queries, transactions, the row-factory trick — then graduates to SQLAlchemy 2.0, the canonical ORM. By the end you'll know when SQLite is enough and when to reach for Postgres.


1. Why SQLite First

SQLite is a full SQL database that lives in a single file on disk. No server, no daemon, no port — sqlite3.connect("app.db") either opens the file or creates it. That's the entire setup.

It's the right database for:

  • Learning SQL without a Docker container in the way
  • Small apps, prototypes, internal tools, mobile apps
  • Test fixtures — spin up an in-memory database with ":memory:"
  • Single-writer workloads where you don't need concurrent writes from many processes

It's the wrong database for:

  • Write-heavy multi-process workloads (writers serialise on a global lock)
  • Anything over ~100 GB or with very high concurrency
  • Multi-server deployments — there's no "remote SQLite"

For everything else there's Postgres. We'll touch on that in Section 10.


2. The Five-Line sqlite3 Tour

python
import sqlite3

conn = sqlite3.connect("app.db")            # opens or creates the file
cur = conn.cursor()
cur.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)")
cur.execute("INSERT INTO users (name) VALUES (?)", ("surya",))
conn.commit()                                # writes to disk

cur.execute("SELECT id, name FROM users")
for row in cur.fetchall():
    print(row)                               # (1, 'surya')

conn.close()

Five concepts:

  • connect() opens (or creates) a database file. ":memory:" for an in-RAM database that vanishes on close — great for tests.
  • cursor() is the handle you run statements against. One connection can have many cursors.
  • execute(sql, params) runs a single statement. executemany(sql, list_of_params) runs the same statement many times — much faster than a Python loop for bulk inserts.
  • fetchall(), fetchone(), or iterating the cursor directly returns rows.
  • commit() flushes pending writes to disk. Without it your INSERTs vanish when the connection closes.

That's a working database in eleven lines. Now we make it safe.


3. Parameterised Queries — The Cardinal Sin

The single most important rule in any database lesson: never build SQL with f-strings or + concatenation. Use ? placeholders and pass values as a tuple.

python
# CATASTROPHIC — SQL injection waiting to happen
name = request.form["name"]                  # user-controlled
cur.execute(f"SELECT * FROM users WHERE name = '{name}'")
+ setup added so this can run · defines request, cur
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

request = _AutoMock('request')
cur = _AutoMock('cur')

If the user types '; DROP TABLE users; -- into your form, the resulting SQL is:

sql
SELECT * FROM users WHERE name = ''; DROP TABLE users; --'

Your users table is now gone. This is the canonical web-app vulnerability — it's been the same for thirty years. The fix is one character of typing:

python
# CORRECT — driver handles escaping; user input is bound as a value, not parsed as SQL
cur.execute("SELECT * FROM users WHERE name = ?", (name,))
+ setup added so this can run · defines cur, name
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

cur = _AutoMock('cur')
name = _AutoMock('name')

The ? is a placeholder. The driver sends the SQL and the values separately, so the database knows the string "'; DROP TABLE users; --" is a name, not a statement. There is no scenario where building SQL with f-strings is acceptable for user-supplied data. None.

Multiple parameters, in order:

python
cur.execute(
    "INSERT INTO books (title, author_id, year) VALUES (?, ?, ?)",
    ("Dune", 1, 1965),
)
+ setup added so this can run · defines cur
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

cur = _AutoMock('cur')

Named parameters work too, with the :name syntax:

python
cur.execute(
    "INSERT INTO books (title, author_id) VALUES (:title, :author)",
    {"title": "Dune", "author": 1},
)
+ setup added so this can run · defines cur
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

cur = _AutoMock('cur')

Both are safe. Pick one and stick with it.


4. Transactions and the with conn: Trick

A transaction groups statements so they either all succeed or all fail. sqlite3 opens an implicit transaction the moment you issue a write — you close it with commit() (save) or rollback() (discard).

python
try:
    cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
except Exception:
    conn.rollback()                          # neither side moved
    raise
+ setup added so this can run · defines cur, conn
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

cur = _AutoMock('cur')
conn = _AutoMock('conn')

There's a cleaner pattern — the connection object is itself a context manager:

python
with conn:                                   # commits on success, rolls back on exception
    conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
+ setup added so this can run · defines conn
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

conn = _AutoMock('conn')

If the block raises, the transaction is rolled back. If it exits cleanly, it's committed. This is the right default — you almost never want a half-applied transfer.

Note: with conn: manages the transaction, not the connection itself. The connection stays open. To close it, call conn.close() or wrap the whole thing in contextlib.closing(conn).


5. sqlite3.Row — Dict-Like Rows

By default, fetchall() returns tuples. Indexing rows by position is brittle — change the SELECT order and every row[2] breaks.

python
conn.row_factory = sqlite3.Row
cur = conn.cursor()
cur.execute("SELECT id, name, email FROM users")
for row in cur:
    print(row["name"], row["email"])         # access by column name
    print(dict(row))                         # convert to a real dict when needed
+ setup added so this can run · defines conn, sqlite3
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

conn = _AutoMock('conn')
sqlite3 = _AutoMock('sqlite3')

Set row_factory once on the connection, and every cursor created from it returns Row objects. They support both row[0] and row["name"], plus dict(row) for serialisation. Two lines of setup for a much cleaner read site.


6. Schema Migrations — The Beginner Version

For real migrations you'll use Alembic (see Section 11). For a small app or a learning project, the version-table pattern is enough:

python
def init_db(conn):
    """Bring the schema up to date. Safe to run on every startup."""
    conn.execute("CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY)")

    cur = conn.execute("SELECT version FROM schema_version")
    row = cur.fetchone()
    current = row[0] if row else 0

    migrations = [
        # v1: initial schema
        """CREATE TABLE authors (
               id INTEGER PRIMARY KEY,
               name TEXT NOT NULL
           )""",
        # v2: books table
        """CREATE TABLE books (
               id INTEGER PRIMARY KEY,
               title TEXT NOT NULL,
               author_id INTEGER NOT NULL REFERENCES authors(id)
           )""",
        # v3: add a year column
        "ALTER TABLE books ADD COLUMN year INTEGER",
    ]

    for v in range(current, len(migrations)):
        conn.execute(migrations[v])
        conn.execute("DELETE FROM schema_version")
        conn.execute("INSERT INTO schema_version (version) VALUES (?)", (v + 1,))
        conn.commit()

CREATE TABLE IF NOT EXISTS is fine for the first run. Once you ship and need to change the schema, the version-table pattern keeps every deployment in lockstep with your code.


7. When SQLite Isn't Enough

SQLite serialises writers — exactly one write transaction can be in flight at a time across all processes. For a personal CLI, a desktop app, or a low-traffic API, that's fine. For a busy web service it's a bottleneck.

Move to PostgreSQL (the default modern choice) or MySQL when you need:

  • Multiple concurrent writers
  • Network access from many app servers
  • Advanced types (JSONB, arrays, full-text search, geospatial)
  • Row-level locking, materialised views, partitioning
  • Database size beyond ~100 GB

Python drivers for production databases:

  • psycopg / psycopg2 — the Postgres driver. psycopg (v3) is the current generation.
  • asyncpg — async Postgres driver, very fast. For asyncio apps.
  • pymysql or mysqlclient — MySQL.

The good news: if you wrote your code against SQLAlchemy (next section), switching from SQLite to Postgres is changing one connection string.


8. SQLAlchemy 2.0 — The Canonical ORM

sqlite3 and other drivers speak SQL directly. SQLAlchemy is the standard Python toolkit one level up — it gives you an ORM (work with Python classes instead of rows) and a Core (compose SQL safely without strings).

bash
pip install sqlalchemy

A complete 2.0-style example:

python
from sqlalchemy import create_engine, ForeignKey, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    books: Mapped[list["Book"]] = relationship(back_populates="author")

class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    author: Mapped[Author] = relationship(back_populates="books")

engine = create_engine("sqlite:///library.db", echo=False)
Base.metadata.create_all(engine)             # creates tables that don't exist

with Session(engine) as session, session.begin():
    frank = Author(name="Frank Herbert")
    frank.books = [Book(title="Dune"), Book(title="Dune Messiah")]
    session.add(frank)
    # auto-commits at the end of session.begin() block

with Session(engine) as session:
    stmt = select(Book).where(Book.title == "Dune")
    for book in session.scalars(stmt):
        print(book.title, "by", book.author.name)

A few concepts worth naming:

  • Engine — the connection pool. Create one per app, share it everywhere.
  • Session — a unit of work. Open one per request/task, close it when done.
  • Declarative models — class Book(Base) with typed Mapped[...] annotations. SQLAlchemy 2.0 uses real type hints, which means your IDE actually helps you.
  • session.begin() — explicit transaction. Commits on success, rolls back on exception. Same pattern as with conn: in stdlib.
  • select(...) — the modern query builder. Returns SQL-safe statement objects.

The connection string "sqlite:///library.db" becomes "postgresql+psycopg://user:pw@host/db" when you switch to Postgres. Everything else stays the same. That portability is the whole point.


9. ORM vs Core — When to Drop Down

The ORM is great when you're working with one or a few objects at a time. For bulk operations or complex analytical queries, drop to SQLAlchemy Core or raw SQL — it's faster and clearer.

python
from sqlalchemy import text

# Bulk insert via Core — much faster than ORM session.add() in a loop
with engine.begin() as conn:
    conn.execute(
        text("INSERT INTO books (title, author_id) VALUES (:title, :author_id)"),
        [
            {"title": "Foundation", "author_id": 2},
            {"title": "I, Robot", "author_id": 2},
            # ...thousands more
        ],
    )
+ setup added so this can run · defines engine
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

engine = _AutoMock('engine')

Rule of thumb: ORM for CRUD on a small number of objects. Core (or raw SQL) for bulk loads, reports, and anything where you'd otherwise be reading the SQL SQLAlchemy generates and wincing.


10. Eager Loading and the N+1 Problem

The classic ORM footgun:

python
# BAD — one query for books, then one query per book for its author. N+1.
for book in session.scalars(select(Book)):
    print(book.title, book.author.name)      # lazy-loads author on each access
+ setup added so this can run · defines session, select, Book
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

session = _AutoMock('session')
def select(*_a, **_kw):
    print('-> select() called')
    return _AutoMock('select()')
Book = _AutoMock('Book')

A page of 50 books fires 51 queries. The fix is to tell SQLAlchemy to fetch the relationship up front:

python
from sqlalchemy.orm import selectinload, joinedload

# GOOD — one query for books, one query for all authors (two queries total)
stmt = select(Book).options(selectinload(Book.author))
for book in session.scalars(stmt):
    print(book.title, book.author.name)
+ setup added so this can run · defines session, select, Book
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

session = _AutoMock('session')
def select(*_a, **_kw):
    print('-> select() called')
    return _AutoMock('select()')
Book = _AutoMock('Book')

selectinload issues a separate IN (...) query for the related rows. joinedload does a SQL JOIN and returns everything in one query — fine for one-to-one, slow for one-to-many because of the row multiplication. Default to selectinload; use joinedload when profiling tells you to.

Eager-load when you know you'll access the relationship. Lazy-load when you might not. Profile with echo=True on the engine to see every query SQLAlchemy generates.


11. Migrations — Alembic in One Paragraph

Once you ship and need to evolve the schema, Alembic is the standard. It's SQLAlchemy's migration tool — alembic revision --autogenerate -m "add year column" produces a migration script by diffing your models against the live schema; alembic upgrade head applies pending migrations. It's the Django-migrations / Rails-migrations equivalent for the SQLAlchemy world. Set it up the day you move past the version-table pattern from Section 6.


12. Common Mistakes

1. String-concatenated SQL. The cardinal sin — see Section 3. f"WHERE id = {user_input}" is a security incident. Use ? placeholders or named bind params. Always.

2. Forgetting to commit. Writes go to the transaction, not the disk, until you commit(). If your script exits without committing, the changes vanish. Use with conn: or with session.begin(): and stop thinking about it.

3. Holding a transaction open across an HTTP request. A transaction holds row-level locks. If you open one when the request arrives and don't commit until the response is built, every other request that touches those rows blocks. Open transactions late, commit early.

4. N+1 queries. See Section 10. The fix is selectinload / joinedload. The detection is echo=True on your engine — if you see one query per row, you have an N+1.

5. Storing JSON-in-text in relational columns. column VARCHAR storing '{"tags": [...]}' is the worst of both worlds — you can't query inside it, the database can't enforce structure, and migrations are painful. Either normalise (a separate tags table) or use a real JSON column (Postgres JSONB, SQLite JSON).

6. Hardcoded credentials. A connection string with user:password@host in source code, committed to git, has leaked production credentials to anyone who clones the repo. Read them from environment variables — see Environment Variables & Configuration.

7. Forgetting to close connections. Use with everywhere — with sqlite3.connect(...) as conn: for stdlib, with Session(engine) as session: for SQLAlchemy. The context managers lesson covers why this matters.


🎯 Your Turn — A Library Database

Build a tiny library schema with authors and books, insert some seed data, and query "all books by a given author" — using only stdlib sqlite3 and parameterised SQL.

Requirements:

1. Two tables — authors(id, name) and books(id, title, author_id REFERENCES authors(id)).
2. Insert three authors: Frank Herbert, Isaac Asimov, Ursula K. Le Guin.
3. Insert five books across those authors.
4. Implement books_by(conn, author_name) returning a list of titles for the given author, using a parameterised query (no f-strings).
5. Use sqlite3.Row so you can access columns by name.
6. Wrap the whole thing in with conn: for transaction safety.

Skeleton:

python
import sqlite3

def init_schema(conn):
    # TODO 1: CREATE TABLE IF NOT EXISTS for authors and books
    ...

def seed(conn):
    # TODO 2: insert three authors and five books
    ...

def books_by(conn, author_name):
    # TODO 3: parameterised SELECT joining books and authors
    ...

conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
with conn:
    init_schema(conn)
    seed(conn)

for title in books_by(conn, "Frank Herbert"):
    print(title)
Hint 1 — Foreign keys in SQLite SQLite supports foreign keys but doesn't enforce them by default. Run conn.execute("PRAGMA foreign_keys = ON") after connecting if you want the constraint actually checked.
Hint 2 — The JOIN SELECT b.title FROM books b JOIN authors a ON a.id = b.author_id WHERE a.name = ?. Pass the name as a single-element tuple: (author_name,).
Show full solution
python
import sqlite3


def init_schema(conn):
    conn.execute("PRAGMA foreign_keys = ON")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS authors (
            id   INTEGER PRIMARY KEY,
            name TEXT NOT NULL UNIQUE
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS books (
            id        INTEGER PRIMARY KEY,
            title     TEXT NOT NULL,
            author_id INTEGER NOT NULL REFERENCES authors(id)
        )
    """)


def seed(conn):
    authors = [("Frank Herbert",), ("Isaac Asimov",), ("Ursula K. Le Guin",)]
    conn.executemany("INSERT INTO authors (name) VALUES (?)", authors)

    # Look up generated IDs by name — never assume insert order
    ids = {row["name"]: row["id"] for row in conn.execute("SELECT id, name FROM authors")}

    books = [
        ("Dune", ids["Frank Herbert"]),
        ("Dune Messiah", ids["Frank Herbert"]),
        ("Foundation", ids["Isaac Asimov"]),
        ("I, Robot", ids["Isaac Asimov"]),
        ("A Wizard of Earthsea", ids["Ursula K. Le Guin"]),
    ]
    conn.executemany("INSERT INTO books (title, author_id) VALUES (?, ?)", books)


def books_by(conn, author_name):
    cur = conn.execute(
        """
        SELECT b.title
        FROM books b
        JOIN authors a ON a.id = b.author_id
        WHERE a.name = ?
        ORDER BY b.title
        """,
        (author_name,),
    )
    return [row["title"] for row in cur]


conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row

with conn:
    init_schema(conn)
    seed(conn)

for name in ["Frank Herbert", "Isaac Asimov", "Ursula K. Le Guin"]:
    print(f"{name}:")
    for title in books_by(conn, name):
        print(f"  - {title}")

What you built:

  • A two-table schema with a real foreign-key relationship.
  • An executemany bulk insert — much faster than a Python loop of executes.
  • A parameterised JOIN that's safe against SQL injection. If someone passed "'; DROP TABLE books; --" as author_name, they'd just get zero results — the database treats it as a literal string.
  • sqlite3.Row so the access is row["title"], not row[0].

If you wanted to swap this for Postgres, replace sqlite3.connect(":memory:") with psycopg.connect("postgresql://...") and the ? placeholders with %s. The rest is identical SQL.


What You Learned

  • SQLite ships with Python — sqlite3.connect("app.db") and you're done. Right tool for prototypes, small apps, and tests.
  • Parameterised queries (? or :name) are non-negotiable. Building SQL with f-strings is an injection vulnerability.
  • with conn: auto-commits on success and rolls back on exception. Same pattern with with session.begin(): in SQLAlchemy.
  • sqlite3.Row lets you access columns by name — set it once on the connection.
  • Schema migrations — CREATE TABLE IF NOT EXISTS for v1, the version-table pattern until you outgrow it, then Alembic.
  • SQLAlchemy 2.0 is the canonical ORM. Engine, Session, declarative models with Mapped[...] annotations.
  • ORM vs Core — ORM for CRUD, Core (or raw SQL) for bulk and analytics.
  • N+1 queries — the classic ORM footgun. selectinload / joinedload to fix.
  • Move to Postgres when you need concurrent writers, advanced types, or remote access.
  • Never hardcode credentials — read them from env vars. See envconfig.

Next: CLI Tools with argparse — build proper command-line interfaces with subcommands, type-coerced flags, and --help that doesn't embarrass you.