Databases in Python — sqlite3 and SQLAlchemy with Diagrams

The Story That Explains Databases

A database is organized storage you can query. Python ships with sqlite3 — a full SQL engine embedded in a single file. For anything bigger — Postgres, MySQL, an ORM, migrations — you reach for SQLAlchemy. Together they cover 95% of what Python programs need.

Visual Diagram — How Your Program Reaches the Database

The Three Building Blocks of Any Database

Your First Database — Nothing to Install

import sqlite3

# Creates the file if it doesn't exist; opens it if it does
conn = sqlite3.connect("bookstore.db")
cur  = conn.cursor()

# Schema — declared ONCE. IF NOT EXISTS makes it re-run-safe.
cur.execute("""
    CREATE TABLE IF NOT EXISTS books (
        id     INTEGER PRIMARY KEY AUTOINCREMENT,
        title  TEXT    NOT NULL,
        author TEXT    NOT NULL,
        price  REAL    NOT NULL,
        stock  INTEGER DEFAULT 0
    )
""")

conn.commit()      # save the schema
conn.close()       # always close when done
print("database ready")

INSERT — Adding Rows

Never insert data with f-strings. That's the #1 way SQL injection sneaks in. Use parameterized queries with ? placeholders — the driver escapes everything correctly.

import sqlite3

with sqlite3.connect("bookstore.db") as conn:
    cur = conn.cursor()

    # Single insert — parameterized with ? placeholders
    cur.execute(
        "INSERT INTO books (title, author, price, stock) VALUES (?, ?, ?, ?)",
        ("Dune", "Frank Herbert", 15.99, 12),
    )

    # Bulk insert — much faster than a loop of execute()
    rows = [
        ("1984",          "George Orwell",   9.99,  30),
        ("Brave New World","Aldous Huxley",   11.50, 18),
        ("Foundation",    "Isaac Asimov",    13.75, 22),
        ("Neuromancer",   "William Gibson",  14.20, 8),
    ]
    cur.executemany(
        "INSERT INTO books (title, author, price, stock) VALUES (?, ?, ?, ?)",
        rows,
    )
    conn.commit()

    print(f"inserted {cur.rowcount + 1} books")

SELECT — Asking Questions

import sqlite3

with sqlite3.connect("bookstore.db") as conn:
    # Row factory: return rows as dict-like objects instead of tuples
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()

    # All books
    cur.execute("SELECT id, title, author, price FROM books")
    for row in cur.fetchall():
        print(f"#{row['id']}  {row['title']:20s} £{row['price']:.2f}")

    # With a WHERE filter — parameterized (safe against injection)
    cur.execute("SELECT * FROM books WHERE price < ? AND stock > ?", (14, 0))
    affordable = cur.fetchall()

    # Aggregate query
    cur.execute("SELECT COUNT(*), AVG(price), MAX(stock) FROM books")
    count, avg_price, max_stock = cur.fetchone()
    print(f"total={count}  avg=£{avg_price:.2f}  most-stocked={max_stock}")

UPDATE & DELETE — Changing State

import sqlite3

with sqlite3.connect("bookstore.db") as conn:
    cur = conn.cursor()

    # Reduce stock when a book is sold — WHERE clause is essential!
    cur.execute("UPDATE books SET stock = stock - 1 WHERE id = ?", (1,))
    print(f"rows updated: {cur.rowcount}")

    # Bulk price rise on cheap books
    cur.execute("UPDATE books SET price = price * 1.1 WHERE price < ?", (12,))

    # Delete out-of-stock, obscure books
    cur.execute("DELETE FROM books WHERE stock = 0 AND price < ?", (10,))
    print(f"rows deleted: {cur.rowcount}")

    conn.commit()

Transactions — All-Or-Nothing Changes

A transaction is a group of changes treated as one atomic unit. Either every change lands, or none do. Classic example: transferring money from one account to another must succeed on both sides — or roll back both sides.

import sqlite3

def transfer(from_id: int, to_id: int, amount: float) -> None:
    conn = sqlite3.connect("bank.db")
    try:
        cur = conn.cursor()
        cur.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?",
                    (amount, from_id))
        cur.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?",
                    (amount, to_id))
        conn.commit()               # both updates land together
    except Exception:
        conn.rollback()             # any error → undo BOTH updates
        raise
    finally:
        conn.close()

# Even simpler — context manager auto-commits or auto-rollbacks
with sqlite3.connect("bank.db") as conn:
    conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    # exiting the `with` block auto-commits; any raised exception auto-rollbacks

Multiple Tables — Foreign Keys & Joins

Real data has relationships. A customer places orders; an order lists books. Split into separate tables and connect them with foreign keys.

import sqlite3

with sqlite3.connect("shop.db") as conn:
    conn.execute("PRAGMA foreign_keys = ON")   # enforce FKs (SQLite quirk)

    conn.executescript("""
        CREATE TABLE IF NOT EXISTS customers (
            id    INTEGER PRIMARY KEY,
            name  TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL
        );
        CREATE TABLE IF NOT EXISTS orders (
            id          INTEGER PRIMARY KEY,
            customer_id INTEGER NOT NULL,
            book_title  TEXT    NOT NULL,
            qty         INTEGER NOT NULL,
            FOREIGN KEY (customer_id) REFERENCES customers(id)
        );
    """)

    # Insert customers
    conn.execute("INSERT OR IGNORE INTO customers (id, name, email) VALUES (?, ?, ?)",
                 (1, "Ada", "ada@example.com"))
    conn.execute("INSERT OR IGNORE INTO customers (id, name, email) VALUES (?, ?, ?)",
                 (2, "Grace", "grace@example.com"))

    # Insert orders that reference customers
    conn.executemany("INSERT INTO orders (customer_id, book_title, qty) VALUES (?, ?, ?)",
                     [(1, "Dune", 2), (1, "1984", 1), (2, "Foundation", 3)])

    # JOIN — combine rows from both tables
    rows = conn.execute("""
        SELECT c.name, o.book_title, o.qty
        FROM   orders o
        JOIN   customers c ON c.id = o.customer_id
        ORDER BY c.name
    """).fetchall()

    for name, title, qty in rows:
        print(f"{name:8s}  {title:12s}  x{qty}")

Enter SQLAlchemy — The Professional Way

sqlite3 is perfect for a single script. Real applications need: Postgres or MySQL support, connection pooling, schema migrations, an object mapper, and a query builder that catches typos before runtime. That's SQLAlchemy — the standard Python database toolkit since 2006 and the backend of every major Python framework (FastAPI, Flask, Django-alternatives, and countless internal tools).

Setup

pip install sqlalchemy

SQLAlchemy Core — SQL, But Composable

The Core layer lets you build SQL as Python expressions. Still SQL, but the query builder catches column-name typos and handles escaping automatically.

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Float, select

# A connection URL tells SQLAlchemy what database. Change this one line to move to Postgres.
#   sqlite:///bookstore.db     ← SQLite
#   postgresql+psycopg://user:pw@host/db
#   mysql+pymysql://user:pw@host/db
engine = create_engine("sqlite:///bookstore.db", echo=False)

metadata = MetaData()

books = Table(
    "books", metadata,
    Column("id",     Integer, primary_key=True),
    Column("title",  String,  nullable=False),
    Column("author", String,  nullable=False),
    Column("price",  Float,   nullable=False),
    Column("stock",  Integer, default=0),
)

metadata.create_all(engine)   # CREATE TABLE IF NOT EXISTS behind the scenes

# Insert
with engine.begin() as conn:                # begin() auto-commits on exit
    conn.execute(books.insert(), [
        {"title": "Dune",       "author": "Frank Herbert", "price": 15.99, "stock": 12},
        {"title": "1984",       "author": "George Orwell", "price": 9.99,  "stock": 30},
        {"title": "Foundation", "author": "Isaac Asimov",  "price": 13.75, "stock": 22},
    ])

# SELECT — as a Python expression
with engine.connect() as conn:
    stmt = select(books.c.title, books.c.price)\
             .where(books.c.price < 14)\
             .order_by(books.c.price)
    for row in conn.execute(stmt):
        print(row.title, row.price)

SQLAlchemy ORM — Rows Are Python Objects

The ORM (Object-Relational Mapper) layer lets you treat rows as normal Python objects. Each class is a table. Each instance is a row. Changes are tracked automatically and flushed to the database when you commit.

from sqlalchemy import create_engine, String, Integer, Float, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

# 1) Base class for all models
class Base(DeclarativeBase):
    pass

# 2) A model — one class = one table
class Book(Base):
    __tablename__ = "books"

    id:     Mapped[int]    = mapped_column(Integer, primary_key=True)
    title:  Mapped[str]    = mapped_column(String(200), nullable=False)
    author: Mapped[str]    = mapped_column(String(100), nullable=False)
    price:  Mapped[float]  = mapped_column(Float, nullable=False)
    stock:  Mapped[int]    = mapped_column(Integer, default=0)

    def __repr__(self) -> str:
        return f"Book({self.title!r}, £{self.price})"

# 3) Create the engine and tables
engine = create_engine("sqlite:///bookstore.db")
Base.metadata.create_all(engine)

# 4) Insert — just create Python objects
with Session(engine) as session:
    session.add_all([
        Book(title="Dune",        author="Frank Herbert", price=15.99, stock=12),
        Book(title="1984",        author="George Orwell", price=9.99,  stock=30),
        Book(title="Foundation",  author="Isaac Asimov",  price=13.75, stock=22),
    ])
    session.commit()

# 5) Query — Python style
with Session(engine) as session:
    stmt = select(Book).where(Book.price < 14).order_by(Book.price)
    for book in session.scalars(stmt):
        print(book)                # uses __repr__

    # Update — modify the object, commit
    dune = session.scalar(select(Book).where(Book.title == "Dune"))
    dune.stock -= 1                    # plain Python attribute assignment
    session.commit()                   # SQL UPDATE runs automatically

Relationships — One-to-Many Made Easy

Two related tables — users and their posts — connect naturally in the ORM. You get typed navigation in both directions: user.posts gives you a list; post.author gives you the user.

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

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str]
    email: Mapped[str] = mapped_column(unique=True)

    # A User has many Posts — this line creates the collection
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id:       Mapped[int] = mapped_column(primary_key=True)
    title:    Mapped[str]
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    # Back-reference — post.author gives you the User
    author: Mapped[User] = relationship(back_populates="posts")

engine = create_engine("sqlite:///blog.db")
Base.metadata.create_all(engine)

# Create a user with posts, all in one shot
with Session(engine) as s:
    ada = User(name="Ada", email="ada@x.io", posts=[
        Post(title="Hello, world"),
        Post(title="Learning SQLAlchemy"),
        Post(title="Why relationships rule"),
    ])
    s.add(ada)
    s.commit()

# Navigate BOTH ways — no SQL written by you
with Session(engine) as s:
    ada = s.scalar(select(User).where(User.name == "Ada"))
    print(f"{ada.name} has {len(ada.posts)} posts:")
    for p in ada.posts:
        print(f"  - {p.title}   (by {p.author.name})")

Common Pitfalls (and Fixes)

Real-World Example — A Full Task Tracker

A complete, production-shaped mini-app: users have many tasks, tasks have a status, and we can query, update, and report — all with the ORM, all with typed models.

from __future__ import annotations
from datetime import datetime
from sqlalchemy import create_engine, String, ForeignKey, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str] = mapped_column(String(80))
    tasks: Mapped[list[Task]] = relationship(
        back_populates="owner", cascade="all, delete-orphan"
    )

class Task(Base):
    __tablename__ = "tasks"
    id:        Mapped[int] = mapped_column(primary_key=True)
    title:     Mapped[str] = mapped_column(String(200))
    status:    Mapped[str] = mapped_column(String(20), default="open")
    created:   Mapped[datetime] = mapped_column(default=datetime.utcnow)
    owner_id:  Mapped[int] = mapped_column(ForeignKey("users.id"))
    owner:     Mapped[User] = relationship(back_populates="tasks")

engine = create_engine("sqlite:///tracker.db", echo=False)
Base.metadata.create_all(engine)

# ── Seed data ────────────────────────────────
with Session(engine) as s:
    if not s.scalar(select(User)):        # only on first run
        s.add_all([
            User(name="Ada", tasks=[
                Task(title="Design schema",      status="done"),
                Task(title="Write ORM models",   status="done"),
                Task(title="Add tests",          status="open"),
            ]),
            User(name="Grace", tasks=[
                Task(title="Draft docs",         status="open"),
                Task(title="Ship v1",            status="blocked"),
            ]),
        ])
        s.commit()

# ── Report: open tasks per user ─────────────
with Session(engine) as s:
    stmt = (
        select(User.name, func.count(Task.id).label("n_open"))
        .join(Task, Task.owner_id == User.id)
        .where(Task.status == "open")
        .group_by(User.name)
    )
    print("Open tasks by user:")
    for name, n in s.execute(stmt):
        print(f"  {name:8s} {n} open")

    # Mark one task as done — the ORM detects the change automatically
    t = s.scalar(select(Task).where(Task.title == "Add tests"))
    if t:
        t.status = "done"
        s.commit()
        print(f"\nmarked done: {t.title}")

Golden Rules