Working with Databases

A database is the standard answer to “I need to store more than fits in a dict and still query it fast”. Python's standard library includes sqlite3, which embeds a full SQL database in a single file — no server to install, no ports to open. For local apps, tests, small services, and personal tools it is often all you need.

The API is dbapi-style: conn = sqlite3.connect("app.db"), cur = conn.cursor(), cur.execute(sql, params), conn.commit(). Always pass parameters as a tuple or dict, never by string concatenation — that is how SQL injection happens. conn supports with to auto-commit or roll back on exit.

For anything beyond a couple dozen queries, use an ORM or a lightweight query builder. SQLAlchemy (Core or ORM) is the industry default for Python; peewee and pony are friendlier alternatives; sqlmodel layers pydantic models on top. They give you migrations, richer types, and protection from a long list of footguns.

Three habits save weeks of debugging. First, always use parameters. Second, use transactions: wrap multi-statement writes in with conn:. Third, index the columns you filter on. Indices turn linear scans into logarithmic lookups and are cheap to add once you know the hot paths.

sqlite3 quick tour

Connect, create table, insert with parameters, query, fetch. cur.fetchone() / cur.fetchall() return Python values. Attach conn.row_factory = sqlite3.Row to get dict-like rows.

executemany(sql, rows) is the batch form — 10–100x faster than a Python loop of execute. Use it for inserts.

ORMs and migrations

SQLAlchemy ORM: define a class per table, use a session for reads and writes. alembic handles schema migrations. The boilerplate is small in exchange for a much safer long-term story.

For web apps, reach for sqlmodel (pydantic + SQLAlchemy) or fastapi-friendly stacks. For pure scripts, raw sqlite3 is perfectly fine.

Database tools.

Tool	Purpose
`sqlite3` module	Embedded SQL in the stdlib.
`sqlite3.Row` class	Dict-like row factory.
`SQLAlchemy` library	Full-featured ORM + SQL toolkit.
`alembic` tool	Schema migrations for SQLAlchemy.
`sqlmodel` library	Typed models with SQLAlchemy under the hood.
`conn.executemany` method	Batched insert / update.
`BEGIN / COMMIT` SQL	Transactions are atomic units.
`CREATE INDEX` SQL	Indices accelerate filtered queries.

Working with Databases code example

The script creates a tiny SQLite database, inserts rows safely, queries, and shows a transaction rollback.

# Lesson: Working with Databases
import sqlite3
from pathlib import Path
from tempfile import gettempdir


db_path = Path(gettempdir()) / "people.db"
if db_path.exists():
    db_path.unlink()

conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row

with conn:  # transaction
    conn.execute("""
        CREATE TABLE people (
            id    INTEGER PRIMARY KEY AUTOINCREMENT,
            name  TEXT NOT NULL,
            age   INTEGER NOT NULL,
            city  TEXT NOT NULL
        )
    """)
    conn.executemany(
        "INSERT INTO people(name, age, city) VALUES (?, ?, ?)",
        [
            ("ana", 30, "oslo"),
            ("ben", 25, "oslo"),
            ("cai", 40, "rome"),
        ],
    )
    conn.execute("CREATE INDEX idx_people_city ON people(city)")

# SELECT with parameters
for row in conn.execute("SELECT * FROM people WHERE city = ? ORDER BY age", ("oslo",)):
    print(dict(row))

# Aggregate
cnt = conn.execute("SELECT city, COUNT(*) AS n FROM people GROUP BY city").fetchall()
print("by city:", [(r["city"], r["n"]) for r in cnt])

# Transaction rollback on error
try:
    with conn:
        conn.execute("INSERT INTO people(name, age, city) VALUES (?, ?, ?)", ("dev", 35, "paris"))
        raise RuntimeError("boom")
except RuntimeError:
    pass

still_three = conn.execute("SELECT COUNT(*) FROM people").fetchone()[0]
print("rows after failed tx:", still_three)

conn.close()
db_path.unlink()

Watch each step:

1) `with conn:` turns a block into an atomic transaction — commit on success, rollback on exception.
2) Parameters go as `(?, ?, ?)` with a tuple — never f-string into SQL.
3) `row_factory = sqlite3.Row` yields dict-like rows.
4) `CREATE INDEX` on `city` speeds up the WHERE clause significantly on larger tables.

Add a running total query.

import sqlite3
conn = sqlite3.connect(":memory:")
conn.executescript("""
    CREATE TABLE sales(day TEXT, amount INT);
    INSERT INTO sales VALUES ('2026-01-01', 10),
                             ('2026-01-02', 15),
                             ('2026-01-03', 7);
""")
rows = conn.execute("SELECT day, amount, SUM(amount) OVER(ORDER BY day) AS running FROM sales").fetchall()
for r in rows:
    print(r)

Parameter-binding prevents injection.

import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE t(x TEXT)")
conn.execute("INSERT INTO t VALUES(?)", ("); DROP TABLE t; --",))
rows = conn.execute("SELECT x FROM t").fetchall()
assert len(rows) == 1

Running prints:

{'id': 2, 'name': 'ben', 'age': 25, 'city': 'oslo'}
{'id': 1, 'name': 'ana', 'age': 30, 'city': 'oslo'}
by city: [('oslo', 2), ('rome', 1)]
rows after failed tx: 3

sqlite3 quick tour

ORMs and migrations

Working with Databases code example

Related Resources