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: strA 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 userThis 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 nothingPOST /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_userturns a token back into a user. - Per-user ownership — the server sets
ownerand scopes every query to it, so tasks stay private. - Generic errors — same
401for bad username or password;404for “not found or not yours” — don’t leak information. - Health endpoint — a trivial
/healthfor 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.