Lesson 5 - Guided Project: Auth-Protected API
Welcome to the Auth-Protected Tasks API
This is the sixth version of the Task Manager you’ve grown across the course, and it’s the one that finally puts a lock on the door. Until now anyone could list, create, or delete tasks; in this project every task endpoint demands proof of who’s calling. You’ll combine everything Module 6 taught — bcrypt password hashing, signed JWTs, and a get_current_user dependency — into a single app where users register, log in for a token, and reach the Task endpoints only with a valid token in hand. To show the difference between who you are and what you’re allowed to do, you’ll add an admin-only delete that turns a normal user away with 403.
By the end of this project, you will be able to:
- Build
/registerand/tokenendpoints that hash passwords and issue JWTs - Gate task endpoints behind a
get_current_userdependency - Add role-based authorization with an admin-only action
- Verify the whole flow — register, login, protected access, roles, expiry — with
TestClient
We’ll build it in stages, run each piece, and finish by exercising the complete flow end to end. Let’s start with the security machinery.
Stage 1: Auth Setup — Hashing, Tokens, and the Current User
Every later stage leans on three things: a password hasher, a way to mint and read JWTs, and a dependency that turns a token back into a user. We set all of it up first.
We use pwdlib with its BcryptHasher so passwords are stored only as bcrypt hashes — never in the clear. JWTs are signed with a secret using PyJWT. The SECRET here is a placeholder; in production you’d load it from an environment variable and never commit it.
from datetime import datetime, timedelta, timezone
import jwt
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from pwdlib import PasswordHash
from pwdlib.hashers.bcrypt import BcryptHasher
pwd = PasswordHash((BcryptHasher(),))
SECRET = "CHANGE-ME-IN-PRODUCTION" # load from env in production; never commit
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# In-memory stores. In a real app these would be the Module 5 database.
users = {}
tasks = {}
next_id = 1
def make_token(username, minutes=30):
return jwt.encode(
{"sub": username, "exp": datetime.now(timezone.utc) + timedelta(minutes=minutes)},
SECRET, algorithm=ALGORITHM,
)
def get_current_user(token: str = Depends(oauth2_scheme)):
exc = HTTPException(status_code=401, detail="Could not validate credentials")
try:
username = jwt.decode(token, SECRET, algorithms=[ALGORITHM]).get("sub")
except jwt.InvalidTokenError:
raise exc
user = users.get(username)
if not user:
raise exc
return userThree pieces do the heavy lifting. pwd hashes and verifies passwords. make_token builds a JWT whose sub claim is the username and whose exp claim makes it expire — so a stolen token can’t live forever. get_current_user is the gate: it pulls the token from the Authorization: Bearer header (via oauth2_scheme), decodes and validates it, looks the user up, and returns that user. Anything wrong along the way — a bad signature, an expired token, an unknown user — collapses into the same 401, so the response never hints at which check failed.
Authorization (what a user may do) is one short step beyond authentication. A second dependency wraps get_current_user and checks the user’s role:
def require_admin(user: dict = Depends(get_current_user)):
if user["role"] != "admin":
raise HTTPException(status_code=403, detail="Admin privileges required")
return userBecause require_admin depends on get_current_user, an admin-only endpoint still needs a valid token first (401 if missing), and then the role is checked (403 if not an admin). That 401-then-403 ordering is exactly the distinction between authentication and authorization.
Stage 2: Register and Log In
Now the two endpoints that get a user into the system. /register creates a user — hashing the password and defaulting the role to "user" — and rejects a duplicate username with 400. /token is the OAuth2 login: it reads the username and password from form data via OAuth2PasswordRequestForm, verifies the password against the stored hash, and returns a JWT on success or 401 on failure.
app = FastAPI()
class RegisterIn(BaseModel):
username: str
password: str
role: str = "user"
@app.post("/register", status_code=201)
def register(body: RegisterIn):
if body.username in users:
raise HTTPException(status_code=400, detail="Username already registered")
users[body.username] = {
"username": body.username,
"hashed_password": pwd.hash(body.password),
"role": body.role,
}
return {"username": body.username, "role": body.role}
@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
user = users.get(form.username)
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(form.username), "token_type": "bearer"}Two details are worth pausing on. First, register stores pwd.hash(body.password) — the plain password is never written anywhere, and the response deliberately returns only the username and role, never the hash. Second, login uses a single check for both an unknown username and a wrong password (not user or not pwd.verify(...)), so an attacker can’t tell whether a username exists by probing the login. The response shape — access_token plus token_type: "bearer" — is what OAuth2 clients and the docs “Authorize” button expect.
Stage 3: Protected Task Endpoints and an Admin-Only Delete
With auth in place, gating the task endpoints is almost anticlimactic: each one just declares a dependency on the current user. Listing and creating tasks require any logged-in user; deleting requires an admin.
class TaskIn(BaseModel):
title: str
@app.get("/tasks")
def list_tasks(user: dict = Depends(get_current_user)):
return list(tasks.values())
@app.post("/tasks", status_code=201)
def create_task(body: TaskIn, user: dict = Depends(get_current_user)):
global next_id
task = {"id": next_id, "title": body.title, "owner": user["username"]}
tasks[next_id] = task
next_id += 1
return task
@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int, user: dict = Depends(require_admin)):
if task_id not in tasks:
raise HTTPException(status_code=404, detail="Task not found")
del tasks[task_id]
return NoneThe pattern is the whole point. Adding user: dict = Depends(get_current_user) to a handler is all it takes to require a valid token; FastAPI runs the dependency first, so an unauthenticated request never reaches your code. Notice that create_task records user["username"] as the task’s owner — the current user isn’t just a gatekeeper, it’s data your endpoint can use. And delete_task swaps in require_admin, so the same token that lets a normal user list and create tasks is not enough to delete one.
Production notes
Three things to carry into a real app. Load SECRET from an environment variable (and rotate it) — never hard-code or commit it. Back users and tasks with the database from Module 5 instead of in-memory dicts, so accounts and tasks survive a restart; the endpoint code barely changes, only where you read and write. And remember the golden rule shown here: passwords are only ever stored as hashes, and your responses never expose them.
Stage 4: Verify the Full Flow
The real test of an auth system is the whole sequence, not any single endpoint. We’ll drive it with TestClient, which calls the app in-process — no server needed. We register a normal user and an admin, log in, hit the protected routes with and without a token, try the admin delete from both roles, and finish with an expired token.
from fastapi.testclient import TestClient
client = TestClient(app)
def show(label, resp):
body = resp.json() if resp.content else ""
print(label, "->", resp.status_code, body)
# Register
show("register ada (user)", client.post("/register", json={"username": "ada", "password": "s3cret"}))
show("register grace (admin)", client.post("/register", json={"username": "grace", "password": "pw", "role": "admin"}))
show("register ada again", client.post("/register", json={"username": "ada", "password": "x"}))
# Login
show("login ada good", client.post("/token", data={"username": "ada", "password": "s3cret"}))
show("login ada bad", client.post("/token", data={"username": "ada", "password": "wrong"}))register ada (user) -> 201 {'username': 'ada', 'role': 'user'}
register grace (admin) -> 201 {'username': 'grace', 'role': 'admin'}
register ada again -> 400 {'detail': 'Username already registered'}
login ada good -> 200 {'access_token': 'eyJhbGciOiJIUzI1NiI...', 'token_type': 'bearer'}
login ada bad -> 401 {'detail': 'Incorrect username or password'}Registration works, the duplicate is rejected with 400, a good login returns a JWT (truncated above), and a wrong password is turned away with 401. Now grab both tokens and exercise the protected endpoints:
ada_token = client.post("/token", data={"username": "ada", "password": "s3cret"}).json()["access_token"]
grace_token = client.post("/token", data={"username": "grace", "password": "pw"}).json()["access_token"]
ada_h = {"Authorization": f"Bearer {ada_token}"}
grace_h = {"Authorization": f"Bearer {grace_token}"}
show("GET /tasks no token", client.get("/tasks"))
show("GET /tasks with token", client.get("/tasks", headers=ada_h))
show("POST /tasks as ada", client.post("/tasks", json={"title": "Write tests"}, headers=ada_h))
show("GET /tasks with token", client.get("/tasks", headers=ada_h))GET /tasks no token -> 401 {'detail': 'Not authenticated'}
GET /tasks with token -> 200 []
POST /tasks as ada -> 201 {'id': 1, 'title': 'Write tests', 'owner': 'ada'}
GET /tasks with token -> 200 [{'id': 1, 'title': 'Write tests', 'owner': 'ada'}]The gate holds: no token gets 401 Not authenticated (from OAuth2PasswordBearer, before our code runs), while a valid token reaches the endpoint, creates a task owned by ada, and lists it back. Finally, role-based authorization and token expiry:
show("DELETE as ada (user)", client.delete("/tasks/1", headers=ada_h))
show("DELETE as grace (admin)", client.delete("/tasks/1", headers=grace_h))
expired = make_token("ada", minutes=-1)
show("GET /tasks expired token", client.get("/tasks", headers={"Authorization": f"Bearer {expired}"}))DELETE as ada (user) -> 403 {'detail': 'Admin privileges required'}
DELETE as grace (admin) -> 204
GET /tasks expired token -> 401 {'detail': 'Could not validate credentials'}Every behavior we wanted is confirmed. The normal user has a perfectly valid token but is forbidden from deleting (403) — authentication succeeded, authorization failed. The admin’s identical-shaped token does let the delete through (204 No Content). And an expired token, even with a correct signature, is rejected with 401, because get_current_user lets jwt.decode reject the stale exp and folds it into the credentials error. That’s the complete loop: who you are, whether you’re logged in, and what your role permits.
To try it interactively, save the app to main.py and run it:
uvicorn main:app --reloadOpen http://127.0.0.1:8000/docs, click Authorize, and enter a username and password you registered. The docs call /token, store the JWT, and attach it to every protected request — so you can exercise the whole flow from the browser, watching the same 200, 401, and 403 responses you just saw in code.
Practice Exercises
Exercise 1: Scope tasks to their owner
Right now GET /tasks returns everyone’s tasks. Change list_tasks so each user sees only the tasks they created, using the owner field already stored on each task.
Hint
You already have the current user inside the endpoint. Filter the returned list with a comprehension: return [t for t in tasks.values() if t["owner"] == user["username"]]. For deletes and updates, also check ownership (or admin) before acting, so one user can’t touch another’s task.
Exercise 2: Add a refresh token
A 30-minute access token forces frequent re-logins. Add a longer-lived refresh token at login and a POST /refresh endpoint that accepts it and returns a fresh access token.
Hint
Issue two tokens at login: a short access token and a longer one (say, days) carrying a claim like {"type": "refresh"}. In /refresh, decode the refresh token, confirm its type is refresh (reject access tokens), look up the user, and mint a new access token with make_token. Keep both signed with the same secret.
Exercise 3: Store users in SQLite
Swap the in-memory users dict for the database you built in Module 5 so accounts survive a restart.
Hint
Create a users table with username, hashed_password, and role columns. Replace users[name] = {...} in register with an INSERT (catch the unique-constraint error to return your 400), and replace users.get(name) in get_current_user and login with a SELECT. The hashing and JWT logic don’t change at all — only where the user comes from.
Summary
You assembled the full Module 6 picture into one app: users register with a bcrypt-hashed password, log in at /token to receive a signed JWT, and reach the Task endpoints only by passing that token through a get_current_user dependency. Layering a require_admin dependency on top added role-based authorization, so a valid login still isn’t enough to delete a task unless you’re an admin. You verified the entire flow with TestClient and saw every guard fire: duplicate registration, bad login, missing token, wrong role, and expired token each produced exactly the right status code.
Key Concepts
- Register endpoint — create a user, hash the password, default the role; reject duplicates with
400. /tokenlogin — verify the password against the stored hash, return a signed JWT, or401.get_current_user— the dependency that decodes a JWT and gates protected endpoints.require_admin— a dependency layered on the current user for role-based authorization (403).- End-to-end verification — the meaningful test is the whole flow, not any single endpoint.
Why This Matters
This is what a real, secured API looks like: an open front door for signing up and logging in, and locked rooms behind it that check both who you are and what you may do. The exact pattern — hash on register, verify and sign a token on login, decode it in a dependency, and add a role check for privileged actions — transfers directly to production services, where you’d swap the in-memory stores for a database and load the secret from the environment. You can now describe and build authentication from scratch, which is among the most commonly tested skills for backend roles.
Next Steps
Continue to Module 7 - Async, Background Work, and Streaming
Move from security to performance: handle concurrency with async endpoints, offload slow work to background tasks, and stream responses to clients.
Back to Module Overview
Return to the Authentication and Security module overview
Continue Building Your Skills
You’ve built a complete auth-protected API — registration, login, token-gated endpoints, and an admin-only action — and proven it works end to end. With security handled, the next module shifts to making your API fast and responsive: running work concurrently, pushing slow jobs into the background, and streaming results as they’re ready.