Mathew K AnalyticsLesson 18 · FastAPI deep dive
FastAPI Tutorial #18: Capstone — Production-Ready API
The final video of the eighteen-part series: everything combined into one real working API. Routers, validation, JWT auth, a real database, custom errors,…
- CourseFastAPI deep dive
- Lesson18 of 18
- FormatJupyter notebook · 10 code cells
What you'll learn
- The Plan - a Small, Real Task Manager API
- The Database Layer - Users and Tasks
- The Schemas - Validation and Response Shapes
- Password Hashing and JWT Helpers
- getcurrentuser - JWT Plus a Real Database Lookup
- The Auth Router - Signup and Login
- The Tasks Router - Create and List, Scoped Per User
- Single-Task Endpoints - Ownership, Update, and Delete
Data
No separate download needed — the notebook creates or downloads everything it uses.
📓 Full notebook
Download .ipynbFastAPI Deep-Dive, Video 18: Capstone - End-to-End Production API#
- The final video of the eighteen-part series: everything combined into one real working API.
- Routers, validation, JWT auth, a real database, custom errors, all in a small task manager.
- Let's build it.
Part 1: The Plan - a Small, Real Task Manager API#
from fastapi import FastAPI, Depends, HTTPException, APIRouter, status, Request
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.testclient import TestClient
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, sessionmaker, Session
from sqlalchemy.pool import StaticPool
from passlib.context import CryptContext
import jwt
from datetime import datetime, timedelta, timezone
app = FastAPI(title='Capstone Task Manager API')
print(app.title)
Part 2: The Database Layer - Users and Tasks#
engine = create_engine('sqlite:///:memory:', connect_args={'check_same_thread': False}, poolclass=StaticPool)
Base = declarative_base()
class UserModel(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String, unique=True, nullable=False)
hashed_password = Column(String, nullable=False)
class TaskModel(Base):
__tablename__ = 'tasks'
id = Column(Integer, primary_key=True)
title = Column(String, nullable=False)
done = Column(Integer, default=0)
owner_id = Column(Integer, ForeignKey('users.id'), nullable=False)
Base.metadata.create_all(bind=engine)
SessionLocal = sessionmaker(bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
print(sorted(Base.metadata.tables.keys()))
Part 3: The Schemas - Validation and Response Shapes#
class UserCreate(BaseModel):
username: str = Field(min_length=3, max_length=30)
password: str = Field(min_length=8)
class UserOut(BaseModel):
id: int
username: str
class Token(BaseModel):
access_token: str
token_type: str
class TaskCreate(BaseModel):
title: str = Field(min_length=1, max_length=200)
class TaskUpdate(BaseModel):
title: str | None = None
done: bool | None = None
class TaskOut(BaseModel):
id: int
title: str
done: bool
print('schemas ready')
Part 4: Password Hashing and JWT Helpers#
pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
SECRET_KEY = 'capstone-demo-secret-key'
ALGORITHM = 'HS256'
def create_access_token(username: str, minutes: int = 30):
expire = datetime.now(timezone.utc) + timedelta(minutes=minutes)
return jwt.encode({'sub': username, 'exp': expire}, SECRET_KEY, algorithm=ALGORITHM)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='auth/login')
print('helpers ready')
Part 5: get_current_user - JWT Plus a Real Database Lookup#
def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username = payload['sub']
except jwt.PyJWTError:
raise HTTPException(status_code=401, detail='could not validate credentials')
user = db.query(UserModel).filter(UserModel.username == username).first()
if not user:
raise HTTPException(status_code=401, detail='user no longer exists')
return user
print('get_current_user ready')
Part 6: The Auth Router - Signup and Login#
auth_router = APIRouter(prefix='/auth', tags=['auth'])
@auth_router.post('/signup', response_model=UserOut, status_code=status.HTTP_201_CREATED)
def signup(payload: UserCreate, db: Session = Depends(get_db)):
existing = db.query(UserModel).filter(UserModel.username == payload.username).first()
if existing:
raise HTTPException(status_code=409, detail='username already taken')
user = UserModel(username=payload.username, hashed_password=pwd_context.hash(payload.password))
db.add(user)
db.commit()
db.refresh(user)
return UserOut(id=user.id, username=user.username)
@auth_router.post('/login', response_model=Token)
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
user = db.query(UserModel).filter(UserModel.username == form_data.username).first()
if not user or not pwd_context.verify(form_data.password, user.hashed_password):
raise HTTPException(status_code=401, detail='incorrect username or password')
token = create_access_token(user.username)
return Token(access_token=token, token_type='bearer')
print('auth router ready')
Part 7: The Tasks Router - Create and List, Scoped Per User#
tasks_router = APIRouter(prefix='/tasks', tags=['tasks'])
@tasks_router.post('', response_model=TaskOut, status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate, current_user: UserModel = Depends(get_current_user), db: Session = Depends(get_db)):
task = TaskModel(title=payload.title, owner_id=current_user.id)
db.add(task)
db.commit()
db.refresh(task)
return TaskOut(id=task.id, title=task.title, done=bool(task.done))
@tasks_router.get('', response_model=list[TaskOut])
def list_tasks(current_user: UserModel = Depends(get_current_user), db: Session = Depends(get_db)):
tasks = db.query(TaskModel).filter(TaskModel.owner_id == current_user.id).all()
return [TaskOut(id=t.id, title=t.title, done=bool(t.done)) for t in tasks]
print('list/create ready')
Part 8: Single-Task Endpoints - Ownership, Update, and Delete#
class TaskNotFoundError(Exception):
pass
def get_owned_task(task_id: int, current_user: UserModel, db: Session):
task = db.get(TaskModel, task_id)
if not task or task.owner_id != current_user.id:
raise TaskNotFoundError()
return task
@tasks_router.get('/{task_id}', response_model=TaskOut)
def get_task(task_id: int, current_user: UserModel = Depends(get_current_user), db: Session = Depends(get_db)):
task = get_owned_task(task_id, current_user, db)
return TaskOut(id=task.id, title=task.title, done=bool(task.done))
@tasks_router.patch('/{task_id}', response_model=TaskOut)
def update_task(task_id: int, payload: TaskUpdate, current_user: UserModel = Depends(get_current_user), db: Session = Depends(get_db)):
task = get_owned_task(task_id, current_user, db)
if payload.title is not None:
task.title = payload.title
if payload.done is not None:
task.done = int(payload.done)
db.commit()
db.refresh(task)
return TaskOut(id=task.id, title=task.title, done=bool(task.done))
@tasks_router.delete('/{task_id}', status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, current_user: UserModel = Depends(get_current_user), db: Session = Depends(get_db)):
task = get_owned_task(task_id, current_user, db)
db.delete(task)
db.commit()
print('detail endpoints ready')
Part 9: Wiring It Together - Error Handling and include_router#
@app.exception_handler(TaskNotFoundError)
async def task_not_found_handler(request: Request, exc: TaskNotFoundError):
return JSONResponse(status_code=404, content={'detail': 'task not found'})
app.include_router(auth_router)
app.include_router(tasks_router)
client = TestClient(app)
print(sorted(app.openapi()['paths'].keys()))
Part 10: The Full Flow - Real Users, Real Isolation, Real Errors#
signup_resp = client.post('/auth/signup', json={'username': 'alex', 'password': 'hunter22'})
print(signup_resp.status_code, signup_resp.json())
login_resp = client.post('/auth/login', data={'username': 'alex', 'password': 'hunter22'})
token = login_resp.json()['access_token']
headers = {'Authorization': f'Bearer {token}'}
created = client.post('/tasks', json={'title': 'write capstone lesson'}, headers=headers).json()
print(created)
print(client.get('/tasks', headers=headers).json())
updated = client.patch(f"/tasks/{created['id']}", json={'done': True}, headers=headers).json()
print(updated)
client.post('/auth/signup', json={'username': 'sam', 'password': 'anotherpass'})
sam_login = client.post('/auth/login', data={'username': 'sam', 'password': 'anotherpass'})
sam_headers = {'Authorization': f"Bearer {sam_login.json()['access_token']}"}
print(client.get(f"/tasks/{created['id']}", headers=sam_headers).status_code)
print(client.get('/tasks').status_code)
print(client.post('/auth/signup', json={'username': 'ab', 'password': 'short'}).status_code)
print(client.delete(f"/tasks/{created['id']}", headers=headers).status_code)
Wrap-Up: What You Built#
- A real database layer: two related SQLAlchemy models, a shared in-memory engine, and get_db.
- Dedicated Pydantic schemas for every request body and response, never leaking a raw password.
- Reusable password-hashing and JWT helpers, unchanged from their own dedicated videos.
- get_current_user combining a decoded JWT with a real database lookup for the current user.
- An auth router handling signup and login, each with its own specific, correct error codes.
- A tasks router where every single endpoint is scoped to the currently authenticated user.
- A shared ownership-checking helper that treats missing and not-owned identically, on purpose.
- A dedicated exception handler turning a custom exception into one consistent error shape.
- Every router and handler wired in before the very first real request ever went out.
- That's the full eighteen-part FastAPI Deep-Dive series. Congratulations, and thanks for building along.
Found this useful?
All lessons, notebooks and datasets here are free. If they helped you, a coffee keeps new lessons coming.
