Building with AI – A Developer's Diary

Starlette Day 3 — SQLite, SQLAlchemy, and Alembic

Replace the in-memory list with a real database: define ORM models, set up SQLAlchemy sessions, and manage your schema with Alembic migrations.

NoCo Interactive • Python · Starlette · SQLAlchemy · Alembic • 20–25 minute read •

Days 1 and 2 gave you a real API structure. The one thing still missing is persistence—every restart wipes your data. Today that changes. You'll wire in SQLite as a database, define ORM models with SQLAlchemy, and set up Alembic to manage schema changes over time. By the end, the CRUD API you built on Day 2 will be reading and writing a real database file.

This is where the Day 2 structure pays off. The routes layer barely changes. You're mostly swapping out the service layer—which is exactly the point of separating them.


Before you start — activate your virtual environment

As with every session, make sure your virtual environment is active before running any commands. New terminal window, new activation:

Shell
cd starlette-day1
source venv/bin/activate   # Mac / Linux
# venv\Scripts\activate    # Windows

Your prompt should show (venv). If pip, uvicorn, or alembic commands fail with "not found," the environment isn't active.

What you're replacing

Right now your app stores tasks in a Python list inside task_service.py. That list only lives for the lifetime of the server process—stop Uvicorn and everything you created is gone. Today you replace it with:

  • SQLite — a file-based database that ships with Python. No server to run, no credentials to configure. Your data lives in a single .db file on disk.
  • SQLAlchemy — a Python ORM that lets you define your tables as Python classes and query them with Python instead of raw SQL.
  • Alembic — a migration tool that tracks changes to your database schema over time, so you can evolve your tables without manually editing the database.

If you've used Eloquent and Laravel migrations, this is the same concept: models define your schema, migrations apply changes to the actual database.

Install the packages

From your project root with the virtual environment active:

Shell
pip install sqlalchemy alembic

Starlette and Uvicorn are already installed from Day 1. This adds the two new dependencies you need for today.

Update the project structure

You need a models folder and a database.py file. Run these from your project root:

Shell
mkdir app/models
touch app/models/__init__.py
touch app/models/task.py
touch app/models/note.py
touch app/database.py

After today, your full project structure will look like this:

Shell
starlette-day1/
├── venv/
├── alembic/                 ← created when you run alembic init
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
├── alembic.ini              ← created when you run alembic init
├── app.db                   ← created when you run alembic upgrade head
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py          ← new
│   ├── models/              ← new
│   │   ├── __init__.py
│   │   ├── task.py
│   │   └── note.py
│   ├── routes/
│   │   ├── __init__.py
│   │   └── tasks.py
│   └── services/
│       ├── __init__.py
│       └── task_service.py
└── main.py                  ← your old Day 1 file, unused

Create the database setup

Open app/database.py and add this:

Python
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

DATABASE_URL = "sqlite:///app.db"

engine = create_engine(
    DATABASE_URL,
    echo=True,
    future=True,
)

SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    autocommit=False,
    future=True,
)

Base = declarative_base()

Three things are happening here:

  • create_engine() creates the connection factory. The URL sqlite:///app.db tells SQLAlchemy to use a file-based SQLite database at app.db in your project root. The three slashes mean "relative path."
  • sessionmaker() creates a reusable factory for database sessions. A session is the unit of work—you open one, do your reads and writes, commit, then close it. autocommit=False means nothing is written to the database until you explicitly call db.commit().
  • declarative_base() gives you a base class your ORM models will inherit from. It's what connects your Python classes to actual database tables.
About echo=True: This logs every SQL statement SQLAlchemy generates to your terminal. It's extremely useful while learning—you can see exactly what queries are being run. Turn it off (or remove it) in production; the output is verbose and carries a performance cost.

Create your models

You'll define two tables: tasks and notes. A task can have many notes—a simple one-to-many relationship that demonstrates how SQLAlchemy handles associations.

app/models/task.py

Python
from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.database import Base


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(255), nullable=False)

    notes: Mapped[list["Note"]] = relationship(
        "Note",
        back_populates="task",
        cascade="all, delete-orphan",
    )

app/models/note.py

Python
from sqlalchemy import ForeignKey, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.database import Base


class Note(Base):
    __tablename__ = "notes"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    task_id: Mapped[int] = mapped_column(ForeignKey("tasks.id"), nullable=False)
    body: Mapped[str] = mapped_column(Text, nullable=False)

    task: Mapped["Task"] = relationship("Task", back_populates="notes")

The cascade="all, delete-orphan" on the Task side means that when you delete a task, SQLAlchemy automatically deletes all of its associated notes. Without this, you'd get a foreign key constraint error when trying to delete a task that has notes attached.

app/models/__init__.py

This file makes both models importable from a single location and—critically—ensures they're loaded into memory when Alembic inspects your metadata:

Python
from app.models.task import Task
from app.models.note import Note

__all__ = ["Task", "Note"]
Why this file matters for Alembic: Alembic's autogenerate works by comparing the actual database schema against the metadata it can see in your Python process. If your models aren't imported anywhere that Alembic's env.py touches, they're invisible—and autogenerate produces an empty or incorrect migration. This __init__.py solves that by giving you one import that loads everything.

Initialize Alembic

From your project root, run:

Shell
alembic init alembic

This creates the alembic/ directory and an alembic.ini config file. You don't need to touch most of what's generated—you only need to update two things: the database URL in alembic.ini, and the metadata reference in alembic/env.py.

Configure Alembic for your app

Step 1 — Update alembic.ini

Open alembic.ini and find this line:

Shell
sqlalchemy.url = driver://user:pass@localhost/dbname

Replace it with:

Shell
sqlalchemy.url = sqlite:///app.db

This must match the DATABASE_URL in app/database.py exactly. If they point to different files, Alembic will apply migrations to one file while your running app reads from another—a confusing situation that's worth preventing now.

Step 2 — Update alembic/env.py

Open alembic/env.py. Replace the entire file with the following. The key additions are importing your Base and models, and setting target_metadata:

Python
from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

from app.database import Base
from app.models import Task, Note

config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        compare_type=True,
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

The critical line is target_metadata = Base.metadata. That's what gives Alembic the schema information it needs. The explicit imports of Task and Note ensure those models are loaded into the process before Alembic inspects the metadata—without those imports, autogenerate would see nothing and generate an empty migration.

Generate your first migration

With Alembic configured and your models in place, generate the initial migration:

Shell
alembic revision --autogenerate -m "create tasks and notes tables"

Alembic compares your model metadata against the current database (which doesn't exist yet, so it's empty) and generates a migration file in alembic/versions/. Open that file—it's plain Python and worth reading. You'll see something like this in the upgrade() function:

Python
def upgrade() -> None:
    op.create_table(
        "tasks",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("title", sa.String(length=255), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_tasks_id"), "tasks", ["id"], unique=False)

    op.create_table(
        "notes",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("task_id", sa.Integer(), nullable=False),
        sa.Column("body", sa.Text(), nullable=False),
        sa.ForeignKeyConstraint(["task_id"], ["tasks.id"]),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_notes_id"), "notes", ["id"], unique=False)

There's also a downgrade() function that drops those tables. That's how you roll back a migration if needed. These files are version-controlled alongside your code—anyone who checks out the project can run the same migrations and get an identical database schema.

Apply the migration

Shell
alembic upgrade head

head means "apply all migrations up to the latest one." You can run this command multiple times safely—Alembic tracks which migrations have already been applied in an alembic_version table it creates inside your database, and skips anything that's already been run.

After this command completes, you should have an app.db file in your project root. Confirm it's there:

Shell
ls -la app.db

If you want to inspect the database directly, DB Browser for SQLite is a free GUI tool that lets you browse tables and run queries. It's useful for confirming your schema looks right and verifying that data is actually being written.

Replace the service layer

Now replace app/services/task_service.py entirely. The function signatures change because every function now takes a db session as its first argument:

Python
from sqlalchemy.orm import Session

from app.models.task import Task


def serialize_task(task: Task) -> dict:
    return {
        "id": task.id,
        "title": task.title,
    }


def get_all_tasks(db: Session) -> list[dict]:
    tasks = db.query(Task).order_by(Task.id.asc()).all()
    return [serialize_task(task) for task in tasks]


def get_task(db: Session, task_id: int) -> dict | None:
    task = db.query(Task).filter(Task.id == task_id).first()
    if not task:
        return None
    return serialize_task(task)


def create_task(db: Session, title: str) -> dict:
    task = Task(title=title)
    db.add(task)
    db.commit()
    db.refresh(task)
    return serialize_task(task)


def update_task(db: Session, task_id: int, title: str) -> dict | None:
    task = db.query(Task).filter(Task.id == task_id).first()
    if not task:
        return None

    task.title = title
    db.commit()
    db.refresh(task)
    return serialize_task(task)


def delete_task(db: Session, task_id: int) -> bool:
    task = db.query(Task).filter(Task.id == task_id).first()
    if not task:
        return False

    db.delete(task)
    db.commit()
    return True

A few things in this code are worth understanding:

  • db.add(task) stages the new row. Nothing is written to the database yet.
  • db.commit() writes the staged changes. Without this call, inserts, updates, and deletes won't persist—this is the single most common mistake when starting with SQLAlchemy.
  • db.refresh(task) reloads the object from the database after the commit. After a commit, SQLAlchemy expires the object's attributes—they won't have current values until you either access them (which triggers a new query) or call refresh() explicitly. Calling it here ensures serialize_task gets accurate data, including the auto-generated id on a new row.
  • delete_task now returns a boolTrue if the task was found and deleted, False if not. The route layer will use this to decide between a 200 and a 404.

Update the routes layer

Replace app/routes/tasks.py. The structure is nearly identical to Day 2—the only difference is opening a session with SessionLocal() and passing it through to the service functions:

Python
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from app.database import SessionLocal
from app.services import task_service


async def get_tasks(request):
    with SessionLocal() as db:
        tasks = task_service.get_all_tasks(db)
        return JSONResponse(tasks)


async def get_task(request):
    task_id = int(request.path_params["id"])

    with SessionLocal() as db:
        task = task_service.get_task(db, task_id)

        if not task:
            return JSONResponse({"error": "Task not found"}, status_code=404)

        return JSONResponse(task)


async def create_task(request: Request):
    data = await request.json()
    title = data.get("title")

    if not title:
        return JSONResponse({"error": "Title is required"}, status_code=400)

    with SessionLocal() as db:
        task = task_service.create_task(db, title)
        return JSONResponse(task, status_code=201)


async def update_task(request: Request):
    task_id = int(request.path_params["id"])
    data = await request.json()
    title = data.get("title")

    if not title:
        return JSONResponse({"error": "Title is required"}, status_code=400)

    with SessionLocal() as db:
        task = task_service.update_task(db, task_id, title)

        if not task:
            return JSONResponse({"error": "Task not found"}, status_code=404)

        return JSONResponse(task)


async def delete_task(request):
    task_id = int(request.path_params["id"])

    with SessionLocal() as db:
        deleted = task_service.delete_task(db, task_id)

        if not deleted:
            return JSONResponse({"error": "Task not found"}, status_code=404)

        return JSONResponse({"message": "Deleted"})


routes = [
    Route("/tasks", get_tasks),
    Route("/tasks", create_task, methods=["POST"]),
    Route("/tasks/{id:int}", get_task),
    Route("/tasks/{id:int}", update_task, methods=["PUT"]),
    Route("/tasks/{id:int}", delete_task, methods=["DELETE"]),
]

The with SessionLocal() as db: pattern opens a session at the start of the block and closes it automatically when the block exits—even if an exception is raised. Closing the session returns the underlying connection back to the pool. For a SQLite app this is simple bookkeeping; for a production database with a connection pool it prevents connection leaks.

Update the app entry point

Open app/main.py and add debug=True:

Python
from starlette.applications import Starlette
from app.routes.tasks import routes as task_routes

app = Starlette(debug=True, routes=task_routes)

debug=True gives you more useful error output in the terminal and browser when something goes wrong. Keep it on during development; remove it before deploying anything publicly.

Run the app

Shell
uvicorn app.main:app --reload

If you still have echo=True in database.py, you'll see SQLAlchemy logging SQL statements in your terminal as each request comes in. This is expected and useful for now.

Test your CRUD endpoints

Create a task

Shell
curl -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "First DB task"}'

Get all tasks

Shell
curl http://127.0.0.1:8000/tasks

Get one task

Shell
curl http://127.0.0.1:8000/tasks/1

Update it

Shell
curl -X PUT http://127.0.0.1:8000/tasks/1 \
  -H "Content-Type: application/json" \
  -d '{"title": "Updated DB task"}'

Delete it

Shell
curl -X DELETE http://127.0.0.1:8000/tasks/1

At this point your database is empty—you created one task and then deleted it. Create a couple more so you have something to verify after the restart:

Shell
curl -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Persistent task one"}'

curl -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Persistent task two"}'

Now stop Uvicorn with Ctrl+C and restart it. Hit /tasks again—both tasks are still there. That's the difference between in-memory and a real database.

Add a second migration — proving the workflow

One of the best ways to understand Alembic is to make a real schema change. Let's add a status field to the Task model.

Update app/models/task.py

Python
from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.database import Base


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    status: Mapped[str] = mapped_column(String(50), nullable=False, server_default="open")

    notes: Mapped[list["Note"]] = relationship(
        "Note",
        back_populates="task",
        cascade="all, delete-orphan",
    )
Note on server_default vs default: SQLAlchemy's default applies the default in Python before sending the INSERT. server_default bakes the default into the column definition itself at the database level. For a new column being added to a table that already has rows, server_default is the right choice—it tells the database how to populate existing rows that don't have a value yet.

You also need to update serialize_task in app/services/task_service.py to include the new field, otherwise it won't appear in your API responses:

Python
def serialize_task(task: Task) -> dict:
    return {
        "id": task.id,
        "title": task.title,
        "status": task.status,
    }

Generate and apply the migration

Before running migration commands, stop the Uvicorn server with Ctrl+C if it's still running. SQLite allows only one writer at a time, and having the app holding an open connection while Alembic tries to alter the schema can cause a "database is locked" error.

Shell
alembic revision --autogenerate -m "add status to tasks"
alembic upgrade head

Open the new file in alembic/versions/. You'll see an op.add_column() call adding the status column to the existing tasks table. The downgrade() function will have the corresponding op.drop_column(). This is the core Alembic loop: change your model, generate a migration, apply it. Your database evolves in step with your code, and the history is preserved in version-controlled files.

Common mistakes

Models not imported in env.py. If autogenerate produces an empty migration, Alembic couldn't see your models. Check that you have the explicit imports at the top of env.py. Having target_metadata = Base.metadata isn't enough on its own—the models need to have been imported so they've registered themselves on Base.

Database URL mismatch. If alembic.ini and app/database.py point to different paths, migrations apply to one file and your app reads from another. Double-check that both use exactly sqlite:///app.db.

Forgetting db.commit(). If creates or updates seem to succeed (no errors) but data doesn't persist, a missing commit()` is almost certainly the reason. SQLAlchemy won't write anything until you explicitly commit.

Manually editing the database file. It's tempting to delete app.db and start fresh when something goes wrong. That's fine during development—but if you do it, you'll need to re-run alembic upgrade head to recreate the tables. Once you have real data you care about, let Alembic manage schema changes rather than touching the file directly.

Running migrations from the wrong directory. Alembic looks for alembic.ini in the current directory. If you get a "can't find alembic.ini" error, make sure you're running the command from your project root, not from inside the app/ folder.

What you actually learned

Today's changes touched a lot of files, but the underlying concepts are straightforward:

  • SQLAlchemy has three core pieces you set up once and reuse everywhere: an engine (connection factory), a SessionLocal (session factory), and a Base (the class your models inherit from).
  • ORM models are Python classes. Each class maps to a table. Each class attribute maps to a column. SQLAlchemy translates your Python operations into SQL.
  • The session lifecycle matters. Open a session, do your work, commit, close. The with SessionLocal() as db: pattern handles the open and close automatically.
  • Alembic migrations are plain Python files. They're version-controlled, reversible, and the single source of truth for how your schema has changed over time.
  • The service/route separation paid off. The routes layer barely changed—it just passes a db session down instead of calling functions with no arguments. The service layer absorbed the entire database swap.
You now have a real backend: HTTP routing, business logic, and persistent storage in separate layers that you can evolve independently. That's the foundation every production backend is built on.

Day 4 preview — Middleware and request lifecycle

On Day 4 you'll add middleware to your Starlette app—CORS headers so a browser frontend can talk to your API, request logging, and an introduction to how the middleware stack processes requests before they reach your routes. The database foundation you built today makes authentication middleware the natural next step after that.

Checkpoint before moving on: make sure all five CRUD operations work against the real database, that data survives a server restart, and that alembic upgrade head runs cleanly with no pending migrations. If those three things are true, Day 4 will go smoothly.

← Back to Building with AI – A Developer's Diary