Databases: SQLite and SQLAlchemy
1 · The lesson
readEvery 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
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.
# 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:
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:
# 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:
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:
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).
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:
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.
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:
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. Forasyncioapps.pymysqlormysqlclient— 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).
pip install sqlalchemy
A complete 2.0-style example:
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 typedMapped[...]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 aswith 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.
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:
# 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:
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:
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. Runconn.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
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
executemanybulk insert — much faster than a Python loop ofexecutes. - A parameterised JOIN that's safe against SQL injection. If someone passed
"'; DROP TABLE books; --"asauthor_name, they'd just get zero results — the database treats it as a literal string. sqlite3.Rowso the access isrow["title"], notrow[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 withwith session.begin():in SQLAlchemy.sqlite3.Rowlets you access columns by name — set it once on the connection.- Schema migrations —
CREATE TABLE IF NOT EXISTSfor 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/joinedloadto 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.