PythonMastery
intermediate 22 min read · lesson 4 of 14 in Python How-To

Databases: SQLite and SQLAlchemy

1 · The lesson

read

Almost every real app needs a database sooner or later: 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, row factories), then moves on to SQLAlchemy 2.0, the most widely used Python ORM. By the end you'll know when SQLite is enough and when to move to Postgres.

1. Why SQLite First

SQLite is a full SQL database that lives in a single file on disk. There is no server, no background process and 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: create 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)
  • Very high concurrency, or more data than you want to keep in one file on one disk
  • Multi-server deployments: there's no "remote SQLite"

For everything else there's Postgres. Section 7 covers when to switch.

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:" gives an in-RAM database that disappears when you close it, which is useful 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. It is 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 are lost when the connection closes.

That's a working database in about ten lines. Next, we make it safe.

3. Parameterised Queries: The Most Important Rule

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

python
# Very dangerous: this is open to SQL injection
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 SQL injection, one of the oldest and best-known web-app vulnerabilities. The fix is a small change:

python
# Correct: the driver sends user input as a value, never 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. Never build SQL with f-strings from user-supplied data.

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 with conn:

A transaction groups statements so they either all succeed or all fail. sqlite3 opens a transaction automatically when you run a write statement. You end 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, because the connection object is also 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, because you almost never want a transfer that is only half done.

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 breaks easily: change the SELECT order and every row[2] is wrong.

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 make the code that reads rows much clearer.

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's schema in step with your code.

7. When SQLite Isn't Enough

SQLite runs writes one at a time: only one write transaction can be active at once, 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
  • More data than fits comfortably in one file on one disk

Python drivers for production databases:

  • psycopg / psycopg2: the Postgres driver. psycopg (v3) is the current generation.
  • asyncpg: an 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 mostly a matter of changing one connection string.

8. SQLAlchemy 2.0: The Standard ORM

sqlite3 and other drivers speak SQL directly. SQLAlchemy is the standard Python toolkit one level above them. It gives you an ORM (you work with Python classes instead of rows) and a Core (you build 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)
    # commits automatically at the end of the 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, so your IDE can autocomplete and check them.
  • session.begin(): an 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. The model and query code stays the same, and that is the main reason to use SQLAlchemy.

9. ORM vs Core: When to Go Lower-Level

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

python
from sqlalchemy import text

# Bulk insert with Core is much faster than ORM session.add() in a loop
with engine.begin() as conn:
    asimov_id = conn.execute(text("INSERT INTO authors (name) VALUES ('Isaac Asimov')")).lastrowid
    conn.execute(
        text("INSERT INTO books (title, author_id) VALUES (:title, :author_id)"),
        [
            {"title": "Foundation", "author_id": asimov_id},
            {"title": "I, Robot", "author_id": asimov_id},
            # ...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 the SQL that the ORM generates would be slow or hard to read.

10. Eager Loading and the N+1 Problem

The classic ORM mistake:

python
# Bad: one query for books, then one more query per book (the N+1 problem)
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 and one for all authors (two queries in 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. It works well for many-to-one and one-to-one relationships, but it is slow for one-to-many, because the parent row is repeated for every child row. Default to selectinload, and 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 change 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, and alembic upgrade head applies pending migrations. It does the same job for SQLAlchemy that Django migrations and Rails migrations do for those frameworks. Set it up the day you move past the version-table pattern from Section 6.

12. Common Mistakes

1. String-concatenated SQL. See Section 3. f"WHERE id = {user_input}" is a security hole. Always use ? placeholders or named bind params.

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(): so the commit happens automatically.

3. Holding a transaction open across an HTTP request. A transaction holds locks on what it has changed (rows in Postgres, or the whole database in SQLite). If you open one when the request arrives and don't commit until the response is built, every other request that touches the same data has to wait. Open transactions late, commit early.

4. N+1 queries. See Section 10. The fix is selectinload / joinedload. To find it, set 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": [...]}' has the problems of both approaches: you can't query inside it easily, 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 contextlib.closing(sqlite3.connect(...)) as conn: for stdlib (a plain with conn: only commits, and it doesn't close), 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". Use 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 SQLite to check the constraint.
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 the generated IDs by name. Never assume the 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, which is 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, because the database treats it as a plain string value.
  • 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. Most of the SQL stays the same, but a few parts are SQLite-only: the PRAGMA line, sqlite3.Row, and the way an INTEGER PRIMARY KEY fills itself in. Postgres has its own versions of those.

What You Learned

  • SQLite comes with Python. sqlite3.connect("app.db") is all the setup you need. It is the right tool for prototypes, small apps and tests.
  • Parameterised queries (? or :name) are required, every time. 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 standard Python 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 are the classic ORM mistake. Fix them with selectinload / joinedload.
  • 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, typed flags, and a clear --help.

Found this useful? Share it with a friend who’s learning Python.