Lesson 4 - Capstone: Build the Complete API

Welcome to the Capstone

This is where it all comes together. Across this course you built a Task Manager piece by piece — endpoints, validation, a database, authentication, routers, settings. In this capstone you’ll assemble those pieces into one complete, coherent application: a real Task Manager API where users register, log in, and manage tasks that only they can see. Every concept from the course earns its place here. This is Part 1 — you build the finished app and prove it works end to end. In Part 2 (the next lesson) you’ll give it a real test suite and deploy it.

By the end of this part, you will be able to:

  • Assemble a full FastAPI app from settings, models, a database session, auth, and routers
  • Wire JWT authentication with hashed passwords into a working register/login flow
  • Enforce per-user ownership so each user only sees their own tasks
  • Verify the complete application end to end with TestClient

We’ll build in stages, each one a small, readable chunk, then assemble and run the whole thing. Let’s begin.


Stage 1: Settings

Every real app needs configuration that lives outside the code — a secret key, a token lifetime, a database URL. You learned to manage these with pydantic-settings: define a Settings class, give each field a default, and let it read overrides from environment variables (or a .env file) automatically.

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    secret_key: str = "CHANGE-ME-IN-PRODUCTION-USE-A-LONG-RANDOM-SECRET"
    access_token_minutes: int = 30
    database_url: str = "sqlite:///app.db"
    model_config = SettingsConfigDict(env_prefix="APP_", env_file=".env")

settings = Settings()

The defaults make the app runnable out of the box, but nothing sensitive is hard-coded for production. With env_prefix="APP_", setting APP_SECRET_KEY in the environment overrides secret_key; the placeholder default is a loud reminder to supply a real, long, random secret before you deploy. We read everything else through settings too, so there’s a single source of truth for configuration.


Stage 2: The Database Models

The app stores two kinds of things: users and tasks. With SQLModel, one class is both your Pydantic schema and your database table. We define a User table and, for tasks, a small family of models so the API can validate input and shape output cleanly.

from sqlmodel import SQLModel, Field

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    username: str = Field(unique=True, index=True)
    hashed_password: str
    role: str = "user"

class TaskBase(SQLModel):
    title: str = Field(min_length=1, max_length=100)
    done: bool = False

class Task(TaskBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    owner: str

class TaskCreate(TaskBase):
    pass

class TaskRead(TaskBase):
    id: int
    owner: str

A User stores a hashed password — never the plain text — plus a role we’ll keep for later (it defaults to "user"). The task models follow the pattern you learned for separating concerns: TaskBase holds the shared fields with validation (a title that must be 1–100 characters), Task is the table (adding the database id and the owner), TaskCreate is what a client sends (just the base fields), and TaskRead is what the API returns. Clients never set id or owner themselves — the server controls those.


Stage 3: The Database Session and Authentication

Now the engine that connects to the database, and the security layer that proves who’s calling. We create the engine from the configured URL, provide a session dependency so every endpoint gets a clean session, and build the JWT pieces: password hashing with pwdlib/bcrypt, a token factory, and the get_current_user dependency that decodes a token into a real user.

from datetime import datetime, timedelta, timezone
import jwt
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from pwdlib import PasswordHash
from pwdlib.hashers.bcrypt import BcryptHasher
from sqlmodel import Session, create_engine, select

pwd = PasswordHash((BcryptHasher(),))
ALGORITHM = "HS256"
engine = create_engine(settings.database_url, connect_args={"check_same_thread": False})

def get_session():
    with Session(engine) as s:
        yield s

oauth2 = OAuth2PasswordBearer(tokenUrl="auth/token")

def make_token(username):
    payload = {
        "sub": username,
        "exp": datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_minutes),
    }
    return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM)

def get_current_user(token: str = Depends(oauth2), session: Session = Depends(get_session)):
    exc = HTTPException(status_code=401, detail="Could not validate credentials")
    try:
        username = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM]).get("sub")
    except jwt.InvalidTokenError:
        raise exc
    user = session.exec(select(User).where(User.username == username)).first()
    if not user:
        raise exc
    return user

This is the heart of the security model. pwd.hash(...) turns a password into a bcrypt hash for storage; pwd.verify(...) checks a login attempt without ever decrypting anything. make_token signs a JWT containing the username (sub) and an expiry (exp), signed with the configured secret. get_current_user reverses that on each protected request: it reads the Bearer token, decodes and verifies the signature, loads the matching user from the database, and raises 401 if anything is off — an invalid signature, an expired token, or a user that no longer exists. Any endpoint that depends on it is automatically protected, and receives the live User object.


Stage 4: The Routers

With the building blocks ready, we group endpoints into APIRouters by concern — one for authentication, one for tasks — exactly as you learned to keep a growing app organized.

The auth router handles registration and login. Both accept the standard OAuth2 form (username and password fields), so they plug straight into the interactive docs’ “Authorize” button.

from fastapi import APIRouter
from fastapi.security import OAuth2PasswordRequestForm

auth_router = APIRouter(prefix="/auth", tags=["auth"])

@auth_router.post("/register", status_code=201)
def register(form: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(get_session)):
    if session.exec(select(User).where(User.username == form.username)).first():
        raise HTTPException(status_code=400, detail="Username already registered")
    user = User(username=form.username, hashed_password=pwd.hash(form.password))
    session.add(user)
    session.commit()
    return {"username": form.username}

@auth_router.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(get_session)):
    user = session.exec(select(User).where(User.username == form.username)).first()
    if not user or not pwd.verify(form.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Incorrect username or password")
    return {"access_token": make_token(user.username), "token_type": "bearer"}

register rejects duplicate usernames and stores only the hash. login verifies the password against that hash and, on success, hands back a signed token. Notice we return the same generic 401 whether the username or the password is wrong — we don’t leak which one.

The tasks router is fully protected: every endpoint depends on get_current_user, and every query is scoped to the current user with Task.owner == user.username. That single filter is what makes tasks private.

tasks_router = APIRouter(prefix="/tasks", tags=["tasks"])

@tasks_router.post("", response_model=TaskRead, status_code=201)
def create_task(data: TaskCreate, user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    task = Task(title=data.title, done=data.done, owner=user.username)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

@tasks_router.get("", response_model=list[TaskRead])
def list_tasks(user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    return session.exec(select(Task).where(Task.owner == user.username)).all()

@tasks_router.delete("/{task_id}", status_code=204)
def delete_task(task_id: int, user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task or task.owner != user.username:
        raise HTTPException(status_code=404, detail="Task not found")
    session.delete(task)
    session.commit()

When creating a task, the server sets owner from the authenticated user — the client can’t claim someone else’s name. When listing, the query only returns tasks the user owns. When deleting, we treat “not yours” the same as “doesn’t exist” and return 404, so a user can’t even probe whether another user’s task ID exists. Ownership is enforced consistently at every entry point.


Stage 5: Assemble the App and Verify

Finally we create the FastAPI app, add a tiny /health endpoint (handy for deployment checks in Part 2), include both routers, and create the database tables.

from fastapi import FastAPI

app = FastAPI(
    title="Task Manager API",
    version="1.0.0",
    description="A complete, tested, deployable Task Manager API.",
)

@app.get("/health", tags=["meta"])
def health():
    return {"status": "ok"}

app.include_router(auth_router)
app.include_router(tasks_router)
SQLModel.metadata.create_all(engine)

That’s the whole application. Here it is verified end to end with TestClient — register a user, log in for a token, create and list tasks, confirm protection and per-user isolation, and check health (the token is truncated for safety):

from fastapi.testclient import TestClient

client = TestClient(app)

# register, then a duplicate
print(client.post("/auth/register", data={"username": "ada", "password": "wonderland"}).status_code)
print(client.post("/auth/register", data={"username": "ada", "password": "wonderland"}).status_code)

# log in and grab the token
token = client.post("/auth/token", data={"username": "ada", "password": "wonderland"}).json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}

# create and list a task
client.post("/tasks", json={"title": "Write the report"}, headers=headers)
print(client.get("/tasks", headers=headers).json())

# protection and isolation
print(client.get("/tasks").status_code)            # no token
client.post("/auth/register", data={"username": "alan", "password": "enigma"})
t2 = client.post("/auth/token", data={"username": "alan", "password": "enigma"}).json()["access_token"]
print(client.get("/tasks", headers={"Authorization": f"Bearer {t2}"}).json())  # alan sees nothing
POST /auth/register     -> 201 {'username': 'ada'}
POST /auth/register dup -> 400 {'detail': 'Username already registered'}
POST /auth/token        -> 200 {'access_token': 'eyJhbG...', 'token_type': 'bearer'}
POST /tasks             -> 201 {'title': 'Write the report', 'done': False, 'id': 1, 'owner': 'ada'}
GET  /tasks             -> 200 [{'title': 'Write the report', 'done': False, 'id': 1, 'owner': 'ada'}]
GET  /tasks (no token)  -> 401 {'detail': 'Not authenticated'}
GET  /tasks (alan)      -> 200 []
DELETE /tasks/1 (ada)   -> 204
GET  /health            -> 200 {'status': 'ok'}

Everything works as designed. Registration creates a user and rejects duplicates; login returns a real signed token; creating a task stamps it with the right owner; listing returns only that owner’s tasks; a request with no token is rejected with 401; a second user (alan) sees an empty list because the tasks belong to ada; deleting your own task returns 204; and /health confirms the app is alive. This is a complete, coherent API.

This one app uses the whole course — and in production you’d split it into files

Look back at what’s here: pydantic-settings for config, SQLModel for the database, dependency injection for sessions and the current user, JWT auth with hashed passwords, APIRouters for organization, and Pydantic request/response models. Every module of the course shows up. We kept it in one file so you can see the whole thing at once, but in a real project you’d split it: settings.py, models.py, database.py, auth.py, routers/auth.py, routers/tasks.py, and a small main.py that creates the app and includes the routers. The secret comes from the environment — never commit a real one — and the long placeholder default is there to be replaced.


Extend the Project

The app is complete, but a real project is never finished. Try these extensions — each builds on a concept the course already gave you.

Exercise 1: Add a PATCH to toggle “done”

Right now you can create, list, and delete tasks, but not update one. Add PATCH /tasks/{task_id} that flips a task’s done flag (and returns the updated task), enforcing the same ownership check as delete_task.

Hint

Mirror delete_task: task = session.get(Task, task_id), then if not task or task.owner != user.username: raise HTTPException(404, ...). Set task.done = not task.done, session.add(task); session.commit(); session.refresh(task), and return it with response_model=TaskRead.

Exercise 2: Add an admin-only route

The User model already has a role field defaulting to "user". Add GET /admin/users that lists every username — but only for users whose role == "admin". Reuse get_current_user and check the role inside the endpoint.

Hint

Inside the endpoint: if user.role != "admin": raise HTTPException(status_code=403, detail="Admin access required"). Use 403 (forbidden, you’re authenticated but not allowed), not 401 (not authenticated). You can promote a user by setting role="admin" directly in the database for now.

Exercise 3: Add created/updated timestamps

Give each Task a created_at set when it’s made. Add a created_at: datetime field to the Task table with a default factory, and include it in TaskRead so the API returns it.

Hint

created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) on the Task table, and add created_at: datetime to TaskRead. Use default_factory (a function called per row), not a bare default — otherwise every task would share one frozen timestamp.


Summary

You assembled the complete Task Manager API from everything the course taught, in clear stages: pydantic-settings for configuration, a SQLModel database with User and Task tables, a session dependency and JWT authentication with hashed passwords, and clean APIRouters for auth and tasks. The result enforces per-user ownership — each user only sees and controls their own tasks — and you verified the whole flow end to end with TestClient: register, login, create, list, protection, isolation, and health.

Key Concepts

  • Layered assembly — settings, models, session, auth, routers, app: each a small, testable piece.
  • JWT auth flow — register stores a hash; login returns a signed token; get_current_user turns a token back into a user.
  • Per-user ownership — the server sets owner and scopes every query to it, so tasks stay private.
  • Generic errors — same 401 for bad username or password; 404 for “not found or not yours” — don’t leak information.
  • Health endpoint — a trivial /health for deployment checks.

Why This Matters

This is what a production API actually looks like: configuration out of the code, a real database, authenticated users, authorization that isolates their data, and a clean structure you can grow. Putting the whole stack together — and proving it runs — is the skill that turns a pile of tutorials into something you can ship. In the next lesson you’ll harden it with a proper test suite and take it the final step: deployment.


Next Steps

Continue to Lesson 5 - Capstone: Test and Deploy It

Part 2 of the capstone: write a real test suite for the complete Task Manager and deploy it to the world.

Back to Module Overview

Return to the Testing, Settings, Deployment, and Capstone module overview


Continue Building Your Skills

You’ve built the complete Task Manager API — settings, database, authentication, ownership, and all — and watched it pass every check end to end. That’s a real, coherent application, not a collection of snippets. Next you’ll do what every professional team does before shipping: write an automated test suite that locks in this behavior, then deploy the app so anyone can use it. The capstone finishes there.