Week 11: FastAPI — Security (Auth and JWT), CORS, Testing, and the Todo API Project

Month 3 | Main goal: The student adds registration, login (OAuth2 + JWT) and protected routes to their API, understands CORS, writes tests with pytest, then finishes and pushes the "Todo API" project to GitHub.

Teacher Preparation (read before the week starts)

This is the hardest week of the month. Do not go into class without running the whole project and seeing all the tests pass (16 passed).

1. Read:
   • FastAPI — "Security - First Steps": https://fastapi.tiangolo.com/tutorial/security/first-steps/
   • FastAPI — "OAuth2 with Password (and hashing), Bearer with JWT tokens": https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/ (this is the basis of Sessions 1–2; the current docs use pwdlib (Argon2) and PyJWT — we follow the same approach)
   • FastAPI — "CORS": https://fastapi.tiangolo.com/tutorial/cors/
   • FastAPI — "Testing": https://fastapi.tiangolo.com/tutorial/testing/
   • SQLModel — "Test Applications with FastAPI and SQLModel": https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/
   • If you have time: https://jwt.io (paste a token to see inside it)

2. Install (in the project venv):

   powershell

   Copy

   pip install "fastapi[standard]" sqlmodel "pwdlib[argon2]" pyjwt pytest
   

   fastapi[standard] already includes python-multipart (needed for the login form) and httpx (needed by TestClient).

3. Prepare the project todo_api on the teaching computer by following Sessions 1–2 step by step — then delete it and start again in class, with your files as a reference on the second computer.

4. Terms to understand beforehand: authentication (who are you?), authorization (what are you allowed to do?), hash, salt, token, JWT, Bearer, OAuth2, the Authorization header, origin, CORS, fixture, assert, dependency override.
5. Remember: The student will type practice passwords like siri123. Tell them early on: never use their real passwords in exercises.

Week Goals (by the end of the week the student will be able to...)

1. Tell apart authentication and authorization, and 401 and 403.
2. Explain why passwords are stored as a hash, and use pwdlib to hash and verify.
3. Explain what a JWT is (header.payload.signature), that it isn't secret (it isn't encryption) but can't be forged without the SECRET_KEY.
4. Build POST /auth/register, POST /auth/token (OAuth2 password flow) and GET /auth/me.
5. Protect routes with a get_current_user dependency and make sure a user sees only their own data.
6. Explain CORS and set up CORSMiddleware.
7. Write tests with TestClient + pytest, using a test database (in-memory) through dependency_overrides.
8. Finish, test, and push the complete Todo API project to GitHub with a README.

Session 1 — Password Hashing, JWT, Register and Login

Session flow

1. Homework review (Contacts API)
2. Authentication vs authorization; the "everyone sees everything" problem
3. Hashing: why not store passwords in plain text; a pwdlib demo
4. JWT: structure, a jwt.encode/decode demo, jwt.io
5. The todo_api project: config, User models, security.py, register, token, me
6. In-class exercise
7. Q&A + homework

Explanation for the teacher (read this first)

Authentication = "Who are you?" (login). Authorization = "What are you allowed to do?" (permissions). Status codes:

• 401 Unauthorized — we don't know you (no token / bad token / expired). The name is misleading — it really means "unauthenticated".
• 403 Forbidden — we know you, but you don't have permission.

Hashing. A hash is a one-way function: hash("siri123") → $argon2id$v=19$m=65536,t=3,p=4$.... You can't go backwards to get the password from the hash. Login works like this: we hash the password just typed and compare it with the stored hash (verify). The benefit: if the database is stolen, the thieves don't get the actual passwords.

• Salt: a random piece added before hashing, so two people with the same password get different hashes. Argon2 adds the salt itself and stores it inside the hash string. That is why running hash("siri123") twice gives different results — and verify still works for both.
• Argon2 is a modern recommended algorithm; it is "deliberately slow" to make guessing millions of passwords hard. pwdlib's PasswordHash.recommended() chooses Argon2.
• Don't use hashlib.sha256(password) alone for passwords — it is far too fast and has no salt.

JWT (JSON Web Token) — a long string with 3 parts separated by dots:

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9 . eyJzdWIiOiJhbWluYSIsImV4cCI6MTkwMDAwMDAwMH0 . YhrVsSS_M1pC...
          header                                  payload                               signature

• Header: the algorithm (HS256).
• Payload: the data ("claims") — sub (subject = who the user is, for us the username), exp (expiry time).
• Signature: made with the SECRET_KEY. If someone changes the payload (for example sub to "admin"), the signature won't match → decode rejects it.
• Important: the header and payload are only base64 — not secret. Anyone can read them (show jwt.io). So don't put passwords or secrets in the payload.
• The server doesn't store the token. It receives it on each request, checks the signature and exp, and reads sub. This fits with HTTP being stateless (Week 9).

OAuth2 Password Flow (as in the FastAPI docs) — a standard for how to ask for a token:

1. The client sends POST /auth/token with form data (not JSON!): username=...&password=.... This is why we use OAuth2PasswordRequestForm and it needs python-multipart.
2. The server returns {"access_token": "...", "token_type": "bearer"}. These names must be exactly so, by the standard.
3. For the following requests, the client sets the header: Authorization: Bearer <token>.

OAuth2PasswordBearer(tokenUrl="auth/token") does two things: (a) it reads the Authorization: Bearer ... header and returns the token (or a 401 "Not authenticated" if it is missing); (b) it tells /docs to add an Authorize button with a login form that calls auth/token.

SECRET_KEY: a long random string. Generate it with:

python -c "import secrets; print(secrets.token_hex(32))"

It must not go to GitHub — we read it from an environment variable (os.getenv). If it is shorter than 32 characters, PyJWT gives an InsecureKeyLengthWarning.

Say this to the student

"Authentication is the guard at the gate asking 'who are you? show me your ID'. Authorization is the guard inside saying 'fine, you're an employee, but you can't enter the director's office'. 401 = you haven't shown ID; 403 = you have ID, but no permission."

"Hashing is like grinding cassava into flour. From cassava you always get flour, but from flour you can't get the cassava back. We store only the 'flour'. When you come to log in, we grind your password again and compare the flour."

"A JWT is like a bus ticket with the company's stamp. Anyone can read what is written on it (name, date), but nobody can forge the stamp. The conductor (the server) just looks at the stamp and the date — he doesn't need to phone the office every time."

"Bearer means 'the one carrying it'. Whoever carries the ticket is allowed in — which is why a token is like cash: if you lose it, whoever picks it up can use it. So it has an expiry time."

Type this code live (share screen)

Step 1 — New project.

cd $HOME\Documents
mkdir todo_api
cd todo_api
python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install "fastapi[standard]" sqlmodel "pwdlib[argon2]" pyjwt pytest
mkdir app, app\routers, tests, demo
New-Item app\__init__.py, app\routers\__init__.py
code .

Step 2 — Hashing demo. demo/hash_demo.py:

from pwdlib import PasswordHash

password_hash = PasswordHash.recommended()

h1 = password_hash.hash("siri123")
h2 = password_hash.hash("siri123")

print(h1)
print(h2)
print("Zinafanana?", h1 == h2)
print("verify sahihi:", password_hash.verify("siri123", h1))
print("verify kosa:  ", password_hash.verify("Siri123", h1))

Output (your hashes will be different — they are random):

$argon2id$v=19$m=65536,t=3,p=4$XlCanl8KWxpUeu89rjj+6g$Y5QsXL9rf7SO6uKuNadlArhSenxN2aTz5TYDVGq8pKY
$argon2id$v=19$m=65536,t=3,p=4$0bN6k7K6Q1mG1w0p6q0x3A$1mJ8...
Zinafanana? False
verify sahihi: True
verify kosa:   False

Ask: "Why are the two hashes different yet both accept siri123?" (Salt.)

Step 3 — JWT demo. demo/jwt_demo.py:

import base64
import json
from datetime import datetime, timedelta, timezone

import jwt

SECRET_KEY = "hii-ni-siri-ya-demo-tu-ndefu-ya-kutosha-32+"

payload = {"sub": "amina", "exp": datetime.now(timezone.utc) + timedelta(minutes=30)}
token = jwt.encode(payload, SECRET_KEY, algorithm="HS256")
print("TOKEN:", token)

# 1) Kusoma payload BILA secret — inawezekana! (si siri)
sehemu_ya_kati = token.split(".")[1]
sehemu_ya_kati += "=" * (-len(sehemu_ya_kati) % 4)   # padding ya base64
print("Payload (base64 tu):", json.loads(base64.urlsafe_b64decode(sehemu_ya_kati)))

# 2) Kuthibitisha kwa secret
print("Decode:", jwt.decode(token, SECRET_KEY, algorithms=["HS256"]))

# 3) Secret mbaya
try:
    jwt.decode(token, "siri-nyingine-kabisa-ndefu-ya-kutosha-32+", algorithms=["HS256"])
except jwt.InvalidSignatureError as e:
    print("Secret mbaya:", e)

# 4) Token iliyoisha muda
old = jwt.encode({"sub": "amina", "exp": datetime.now(timezone.utc) - timedelta(minutes=1)},
                 SECRET_KEY, algorithm="HS256")
try:
    jwt.decode(old, SECRET_KEY, algorithms=["HS256"])
except jwt.ExpiredSignatureError as e:
    print("Imeisha muda:", e)

Output (the token will differ):

TOKEN: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhbWluYSIsImV4cCI6MTc5OTk5OTk5OX0.k3...
Payload (base64 tu): {'sub': 'amina', 'exp': 1799999999}
Decode: {'sub': 'amina', 'exp': 1799999999}
Secret mbaya: Signature verification failed
Imeisha muda: Signature has expired

Paste the token into https://jwt.io — show that the payload is readable. Emphasise: don't put secrets in the payload. (InvalidSignatureError and ExpiredSignatureError are both children of jwt.InvalidTokenError — that is the one we will catch in the app.)

Step 4 — app/config.py:

import os

# Kwenye kazi halisi, SECRET_KEY lazima itoke kwenye environment variable,
# isiandikwe kwenye code inayoenda GitHub.
SECRET_KEY = os.getenv("SECRET_KEY", "badilisha-hii-kwa-siri-ndefu-ya-random-angalau-herufi-32")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///todo.db")

Step 5 — app/database.py (like Week 10, but the URL comes from config):

from typing import Annotated

from fastapi import Depends
from sqlmodel import Session, SQLModel, create_engine

from app.config import DATABASE_URL

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})


def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine)


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


SessionDep = Annotated[Session, Depends(get_session)]

Step 6 — app/models.py (the users and token part; we will add tasks in Session 2):

from datetime import datetime, timezone

from sqlmodel import Field, SQLModel


# ---------- Users ----------
class UserBase(SQLModel):
    username: str = Field(min_length=3, max_length=30, unique=True, index=True)
    full_name: str | None = None


class User(UserBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    hashed_password: str


class UserCreate(UserBase):
    password: str = Field(min_length=6, max_length=128)


class UserPublic(UserBase):
    id: int


# ---------- Token ----------
class Token(SQLModel):
    access_token: str
    token_type: str = "bearer"

Explain: UserCreate has password (coming in), User has hashed_password (stored), UserPublic has nothing about passwords (going out). This is the benefit of multiple models we saw in Week 10.

Step 7 — app/security.py:

from datetime import datetime, timedelta, timezone

import jwt
from pwdlib import PasswordHash

from app.config import ACCESS_TOKEN_EXPIRE_MINUTES, ALGORITHM, SECRET_KEY

password_hash = PasswordHash.recommended()   # Argon2


def hash_password(password: str) -> str:
    return password_hash.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)


def create_access_token(subject: str, expires_minutes: int = ACCESS_TOKEN_EXPIRE_MINUTES) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=expires_minutes)
    payload = {"sub": subject, "exp": expire}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)


def decode_access_token(token: str) -> str | None:
    """Inarudisha username (sub) au None kama token si sahihi/imeisha muda."""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.InvalidTokenError:
        return None
    return payload.get("sub")

Step 8 — app/deps.py:

from typing import Annotated

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlmodel import select

from app.database import SessionDep
from app.models import User
from app.security import decode_access_token

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


def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], session: SessionDep) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Token si sahihi au imeisha muda",
        headers={"WWW-Authenticate": "Bearer"},
    )
    username = decode_access_token(token)
    if username is None:
        raise credentials_exception
    user = session.exec(select(User).where(User.username == username)).first()
    if user is None:
        raise credentials_exception
    return user


CurrentUser = Annotated[User, Depends(get_current_user)]

Draw the chain on the board (or in a comment): endpoint ← CurrentUser ← get_current_user ← (oauth2_scheme → token) + (SessionDep → session).

Step 9 — app/routers/auth.py:

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlmodel import select

from app.database import SessionDep
from app.deps import CurrentUser
from app.models import Token, User, UserCreate, UserPublic
from app.security import create_access_token, hash_password, verify_password

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


@router.post("/register", response_model=UserPublic, status_code=status.HTTP_201_CREATED)
def register(user_in: UserCreate, session: SessionDep):
    ipo = session.exec(select(User).where(User.username == user_in.username)).first()
    if ipo:
        raise HTTPException(status_code=400, detail="Username tayari imechukuliwa")
    user = User(
        username=user_in.username,
        full_name=user_in.full_name,
        hashed_password=hash_password(user_in.password),
    )
    session.add(user)
    session.commit()
    session.refresh(user)
    return user


@router.post("/token", response_model=Token)
def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()], session: SessionDep):
    user = session.exec(select(User).where(User.username == form_data.username)).first()
    if user is None or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Username au password si sahihi",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return Token(access_token=create_access_token(user.username))


@router.get("/me", response_model=UserPublic)
def read_me(current_user: CurrentUser):
    return current_user

Emphasise: the login message is the same for "username doesn't exist" and "wrong password" — we don't tell an attacker which one was wrong.

Step 10 — app/main.py:

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import create_db_and_tables
from app.routers import auth


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(title="Todo API", version="1.0.0", lifespan=lifespan)
app.include_router(auth.router)


@app.get("/", tags=["mwanzo"])
def mwanzo():
    return {"ujumbe": "Todo API — nenda /docs"}

Run fastapi dev app/main.py. In /docs:

1. POST /auth/register → {"username": "amina", "password": "siri123", "full_name": "Amina Said"} → 201 {"username":"amina","full_name":"Amina Said","id":1} (no password!)
2. Register the same username again → 400
3. GET /auth/me without logging in → 401 {"detail":"Not authenticated"}
4. Click Authorize 🔒 (top right) → enter amina / siri123 → Authorize → Close
5. GET /auth/me again → 200. Open the "Curl" section in /docs — show the header -H 'Authorization: Bearer eyJ...'. Say: "/docs adds the token to every request by itself."
6. Open todo.db with SQLite Viewer → the user table → the hashed_password column starts with $argon2id$.

Questions to ask the student

1. What is the difference between 401 and 403? — 401: we don't know you (no token/bad token). 403: we know you, you have no permission.
2. Why don't we store passwords in plain text? — If the database leaks, all the passwords leak (and people use one password in many places).
3. If a hash can't be reversed, how does login work? — We hash/verify the new password against the stored hash.
4. If someone steals a JWT, what can they read? Can they change sub? — They can read the payload (it isn't secret). They can't change it without the SECRET_KEY — the signature will reject it. But they can use it as it is until it expires.
5. Why does /auth/token receive form data rather than JSON? — That is what the OAuth2 password flow standard requires; /docs and other tools depend on it.
6. Why is there one login message for two errors? — So that an attacker doesn't learn whether the username exists.
7. What happens if the SECRET_KEY leaks? — Someone can create a token for any user. It must be changed (and all old tokens die).

Common mistakes

In-class exercise

Exercise 1.1 — Using requests (with the server running), write demo/login_client.py which: (a) registers the user baraka/siri456 (if they already exist, just carry on), (b) logs in and gets a token, (c) calls /auth/me and prints the username.

Exercise 1.2 — Change UserBase so the username is only lowercase letters, numbers and _ (hint: Annotated[str, StringConstraints(pattern=r"^[a-z0-9_]+$")] from pydantic, like Simu in Week 10). Try registering "Baraka Juma".

Exercise 1.3 — Open your token on jwt.io. Write down: what is sub? What number is exp? Convert exp to a date with Python.

import requests

BASE = "http://127.0.0.1:8000"

r = requests.post(f"{BASE}/auth/register",
                  json={"username": "baraka", "password": "siri456"}, timeout=5)
print("Register:", r.status_code)   # 201 mara ya kwanza, 400 baadaye

r = requests.post(f"{BASE}/auth/token",
                  data={"username": "baraka", "password": "siri456"}, timeout=5)
r.raise_for_status()
token = r.json()["access_token"]
print("Token:", token[:25], "...")

headers = {"Authorization": f"Bearer {token}"}
me = requests.get(f"{BASE}/auth/me", headers=headers, timeout=5)
print("Mimi ni:", me.json()["username"])

from typing import Annotated

from pydantic import StringConstraints

Username = Annotated[str, StringConstraints(min_length=3, max_length=30, pattern=r"^[a-z0-9_]+$")]


class UserBase(SQLModel):
    username: Username = Field(unique=True, index=True)
    full_name: str | None = None

from datetime import datetime, timezone
print(datetime.fromtimestamp(1799999999, tz=timezone.utc))   # weka exp yako

Session 2 — Protected Routes, CORS, and Testing

Session flow

1. Homework review
2. Tasks with owner_id; protecting routes with CurrentUser; a user sees only their own
3. CORS: a demo of an HTML page on port 5500; the error; CORSMiddleware
4. Testing: TestClient, pytest, fixtures, a test database, dependency_overrides
5. In-class exercise
6. Q&A + homework

Explanation for the teacher (read this first)

Protecting a route is now very easy: add the argument current_user: CurrentUser. If the token is missing/bad, get_current_user raises a 401 and the endpoint doesn't run at all. If it is fine, the endpoint gets a User object.

Ownership. Each Task has an owner_id (a foreign key to user.id). The rules:

• Creating: owner_id = current_user.id — never let the client send an owner_id (that is why it isn't in TaskCreate).
• Reading the list: where(Task.owner_id == current_user.id).
• Reading/changing/deleting one: if the task isn't theirs → we return 404 (not 403). Why? A 403 would tell them "task number 7 exists, but it isn't yours" — we would be leaking information. 404 is safer. (Both are acceptable; we choose 404.)

CORS (Cross-Origin Resource Sharing). Origin = scheme + host + port. http://127.0.0.1:5500 and http://127.0.0.1:8000 are different origins (different ports). The browser, for security, doesn't let JavaScript on a page from origin A read a response from an API on origin B — unless the API says explicitly "I allow origin A" through the Access-Control-Allow-Origin header.

• CORS is a rule of the browser only. requests, curl, /docs (which is on the same origin) aren't affected — that is why the student has never seen it until now.
• For "non-simple" requests (for example those with an Authorization header or Content-Type: application/json), the browser first sends a preflight: an OPTIONS request asking for permission.
• FastAPI's CORSMiddleware answers the preflight and adds the headers. allow_origins should be a specific list; ["*"] allows everyone (fine for experiments, not for production with cookies/credentials).

Testing. Instead of clicking through /docs every time, we write code that tests our API for us.

• TestClient(app) — a "fake" client that talks to the app directly without a real server (no need for fastapi dev). Its API is similar to requests: client.get(...), client.post(..., json=...), data=..., headers=....
• pytest — finds test_*.py files and test_* functions, runs them, and reports. If an assert fails → the test fails.
• Fixture — a function with @pytest.fixture that prepares something (for example a client, or a token). A test asks for it by argument name — this is pytest's dependency injection!
• Test database — we don't want tests to damage our todo.db. We create a new memory database ("sqlite://") for each test, and with app.dependency_overrides[get_session] = ... we tell FastAPI: "instead of the normal get_session, use this one". StaticPool makes sure all connections use the same memory database.
• TestClient(app) without with doesn't run lifespan, so todo.db isn't touched at all.
• Running: pytest needs to know the project folder is on sys.path for from app.main import app to work. Two ways: (a) python -m pytest (adds the current folder), or (b) a pyproject.toml file with pythonpath = ["."], then pytest alone works. We will do (b).

Say this to the student

"Protecting a route is like posting a guard at the door: you just add current_user: CurrentUser and the guard is there. No ticket — no entry."

"Everyone has their own drawer in the office cabinet. Even if you know the number of your colleague's drawer, when you open it we tell you 'that drawer doesn't exist' — we don't tell you 'it exists but it isn't yours'."

"CORS is like a school rule: the parent (the browser) doesn't let the child (the page's JavaScript) accept a gift from a stranger (another origin), until that person shows a letter of permission (Access-Control-Allow-Origin). But you, an adult (requests, curl), have no parent stopping you."

"Tests are like a teacher marking your API's exam every time you change the code — 2 seconds, without clicking anything. If you break something without realising, a test will tell you."

Type this code live (share screen)

Step 1 — Add Tasks to app/models.py (below Token):

# ---------- Tasks ----------
class TaskBase(SQLModel):
    title: str = Field(min_length=1, max_length=100)
    description: str | None = None
    done: bool = False


class Task(TaskBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    owner_id: int = Field(foreign_key="user.id", index=True)
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


class TaskCreate(TaskBase):
    pass


class TaskPublic(TaskBase):
    id: int
    owner_id: int
    created_at: datetime


class TaskUpdate(SQLModel):
    title: str | None = Field(default=None, min_length=1, max_length=100)
    description: str | None = None
    done: bool | None = None

Explain default_factory: a function called each time a new task is created (like field(default_factory=list) in dataclasses). If we wrote default=datetime.now(...), it would be computed only once at import — all tasks would get the same time.

Step 2 — app/routers/tasks.py:

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlmodel import select

from app.database import SessionDep
from app.deps import CurrentUser
from app.models import Task, TaskCreate, TaskPublic, TaskUpdate

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


def get_own_task(task_id: int, session: SessionDep, current_user: CurrentUser) -> Task:
    task = session.get(Task, task_id)
    # Kama task si yako, tunajibu 404 (si 403) ili tusifichue kwamba ipo.
    if task is None or task.owner_id != current_user.id:
        raise HTTPException(status_code=404, detail="Task haipo")
    return task


OwnTask = Annotated[Task, Depends(get_own_task)]


@router.post("/", response_model=TaskPublic, status_code=status.HTTP_201_CREATED)
def create_task(task_in: TaskCreate, session: SessionDep, current_user: CurrentUser):
    task = Task(**task_in.model_dump(), owner_id=current_user.id)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@router.get("/", response_model=list[TaskPublic])
def list_tasks(
    session: SessionDep,
    current_user: CurrentUser,
    done: bool | None = None,
    offset: int = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
    query = select(Task).where(Task.owner_id == current_user.id)
    if done is not None:
        query = query.where(Task.done == done)
    query = query.order_by(Task.id).offset(offset).limit(limit)
    return session.exec(query).all()


@router.get("/{task_id}", response_model=TaskPublic)
def read_task(task: OwnTask):
    return task


@router.patch("/{task_id}", response_model=TaskPublic)
def update_task(task: OwnTask, task_in: TaskUpdate, session: SessionDep):
    task.sqlmodel_update(task_in.model_dump(exclude_unset=True))
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task: OwnTask, session: SessionDep):
    session.delete(task)
    session.commit()

In app/main.py add from app.routers import auth, tasks and app.include_router(tasks.router).

Delete todo.db (new Task model) — stop the server, delete the file, start it again. Then in /docs:

1. Without Authorize: GET /tasks/ → 401. Look — every tasks endpoint has a lock 🔒.
2. Authorize as amina → create 2 tasks → GET /tasks/ → 2.
3. Register baraka, Logout in Authorize, log in as baraka → GET /tasks/ → []. GET /tasks/1 → 404. Say: "This is authorization: Baraka can't see Amina's tasks."

Step 3 — CORS demo. Create frontend/index.html:

<!doctype html>
<html lang="sw">
<head><meta charset="utf-8"><title>Todo Frontend</title></head>
<body>
  <h1>Jaribio la CORS</h1>
  <button id="btn">Ita API</button>
  <pre id="matokeo"></pre>
  <script>
    document.getElementById("btn").onclick = async () => {
      const out = document.getElementById("matokeo");
      try {
        const r = await fetch("http://127.0.0.1:8000/");
        out.textContent = JSON.stringify(await r.json(), null, 2);
      } catch (e) {
        out.textContent = "KOSA: " + e;
      }
    };
  </script>
</body>
</html>

In a SECOND terminal (with the FastAPI server still running in the first):

cd frontend
python -m http.server 5500

Open http://127.0.0.1:5500 → click "Ita API" (call the API) → KOSA: TypeError: Failed to fetch. Open F12 → Console:

Access to fetch at 'http://127.0.0.1:8000/' from origin 'http://127.0.0.1:5500' has been blocked by CORS policy: No 'Access-Control-Allow-Origin' header is present on the requested resource.

Say: "The server answered (look at the FastAPI terminal — 200!), but the browser stopped JavaScript from reading the answer."

Now in app/main.py add:

from fastapi.middleware.cors import CORSMiddleware

# ... baada ya app = FastAPI(...)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5500", "http://127.0.0.1:5500"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Click again → {"ujumbe": "Todo API — nenda /docs"}. In the Network tab → response headers → access-control-allow-origin: http://127.0.0.1:5500. Stop the 5500 server with Ctrl+C.

Step 4 — Testing. Create pyproject.toml in the project root:

[tool.pytest.ini_options]
pythonpath = ["."]
testpaths = ["tests"]

tests/conftest.py (pytest reads it by itself — fixtures in it are available to all tests):

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from app.database import get_session
from app.main import app


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://",                               # database ya memory, mpya kila test
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


@pytest.fixture(name="auth_headers")
def auth_headers_fixture(client: TestClient) -> dict[str, str]:
    client.post("/auth/register", json={"username": "amina", "password": "siri123"})
    r = client.post("/auth/token", data={"username": "amina", "password": "siri123"})
    token = r.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}

tests/test_auth.py:

from fastapi.testclient import TestClient


def test_register_inafanikiwa(client: TestClient):
    r = client.post("/auth/register", json={"username": "juma", "password": "siri123"})
    assert r.status_code == 201
    data = r.json()
    assert data["username"] == "juma"
    assert "password" not in data
    assert "hashed_password" not in data


def test_register_username_iliyopo(client: TestClient):
    client.post("/auth/register", json={"username": "juma", "password": "siri123"})
    r = client.post("/auth/register", json={"username": "juma", "password": "nyingine1"})
    assert r.status_code == 400


def test_register_password_fupi(client: TestClient):
    r = client.post("/auth/register", json={"username": "juma", "password": "123"})
    assert r.status_code == 422


def test_login_password_mbaya(client: TestClient):
    client.post("/auth/register", json={"username": "juma", "password": "siri123"})
    r = client.post("/auth/token", data={"username": "juma", "password": "kosa"})
    assert r.status_code == 401


def test_me_bila_token(client: TestClient):
    r = client.get("/auth/me")
    assert r.status_code == 401


def test_me_na_token(client: TestClient, auth_headers: dict):
    r = client.get("/auth/me", headers=auth_headers)
    assert r.status_code == 200
    assert r.json()["username"] == "amina"


def test_token_bandia(client: TestClient):
    r = client.get("/auth/me", headers={"Authorization": "Bearer abc.def.ghi"})
    assert r.status_code == 401

Run:

pytest -v

Output:

tests/test_auth.py::test_register_inafanikiwa PASSED                     [ 14%]
tests/test_auth.py::test_register_username_iliyopo PASSED                [ 28%]
tests/test_auth.py::test_register_password_fupi PASSED                   [ 42%]
tests/test_auth.py::test_login_password_mbaya PASSED                     [ 57%]
tests/test_auth.py::test_me_bila_token PASSED                            [ 71%]
tests/test_auth.py::test_me_na_token PASSED                              [ 85%]
tests/test_auth.py::test_token_bandia PASSED                             [100%]

============================== 7 passed in 1.20s ==============================

Break something on purpose: in register, change status_code=status.HTTP_201_CREATED to 200, run pytest → FAILED ... assert 200 == 201. Change it back. Say: "This is the job of tests — catching mistakes before a customer sees them."

Also note: todo.db wasn't touched (the tests use the memory database).

Questions to ask the student

1. Why isn't owner_id in TaskCreate? — The client could create a task in someone else's name; the server sets it from current_user.
2. If Baraka requests Amina's /tasks/1, what does he get and why? — 404; we don't want to reveal that the task exists.
3. Why did we never see a CORS error when using requests or /docs? — CORS is a browser rule for JavaScript from a different origin; requests isn't a browser, and /docs is on the same origin.
4. Are http://localhost:5500 and http://127.0.0.1:5500 one origin? — No! Different host names → different origins. That is why we added both.
5. What is a fixture? — A pytest function that prepares something for tests; a test asks for it by argument name.
6. Why do we use a memory database in tests? — It is fast, new for each test (tests don't depend on each other), and doesn't damage real data.
7. What does app.dependency_overrides do? — It replaces a dependency (here get_session) with another during tests.

Common mistakes

In-class exercise

Exercise 2.1 — Write tests/test_tasks.py with these tests:

• (a) creating a task and seeing it in the list
• (b) GET /tasks/ and POST /tasks/ without a token → 401
• (c) PATCH done=true then DELETE → 204, then GET → 404
• (d) a second user doesn't see the first user's task

Exercise 2.2 — Write a test proving that an expired token is rejected (hint: create_access_token("x", expires_minutes=-1)).

Exercise 2.3 — Write a CORS test: client.options("/tasks/", headers={"Origin": "http://127.0.0.1:5500", "Access-Control-Request-Method": "GET"}) and confirm the access-control-allow-origin header.

# tests/test_tasks.py
from fastapi.testclient import TestClient

from app.security import create_access_token


def test_create_na_list(client: TestClient, auth_headers: dict):
    r = client.post("/tasks/", json={"title": "Kusoma"}, headers=auth_headers)
    assert r.status_code == 201
    assert r.json()["done"] is False

    r = client.get("/tasks/", headers=auth_headers)
    assert r.status_code == 200
    assert len(r.json()) == 1


def test_tasks_zinahitaji_login(client: TestClient):
    assert client.get("/tasks/").status_code == 401
    assert client.post("/tasks/", json={"title": "x"}).status_code == 401


def test_patch_na_delete(client: TestClient, auth_headers: dict):
    task_id = client.post("/tasks/", json={"title": "Kufua"}, headers=auth_headers).json()["id"]

    r = client.patch(f"/tasks/{task_id}", json={"done": True}, headers=auth_headers)
    assert r.json()["done"] is True
    assert r.json()["title"] == "Kufua"

    assert client.delete(f"/tasks/{task_id}", headers=auth_headers).status_code == 204
    assert client.get(f"/tasks/{task_id}", headers=auth_headers).status_code == 404


def test_huwezi_kuona_task_ya_mwingine(client: TestClient, auth_headers: dict):
    task_id = client.post("/tasks/", json={"title": "Siri ya Amina"}, headers=auth_headers).json()["id"]

    client.post("/auth/register", json={"username": "baraka", "password": "siri456"})
    token = client.post("/auth/token", data={"username": "baraka", "password": "siri456"}).json()["access_token"]
    baraka = {"Authorization": f"Bearer {token}"}

    assert client.get(f"/tasks/{task_id}", headers=baraka).status_code == 404
    assert client.get("/tasks/", headers=baraka).json() == []


# 2.2
def test_token_iliyoisha_muda(client: TestClient):
    client.post("/auth/register", json={"username": "zawadi", "password": "siri123"})
    old = create_access_token("zawadi", expires_minutes=-1)
    r = client.get("/auth/me", headers={"Authorization": f"Bearer {old}"})
    assert r.status_code == 401


# 2.3
def test_cors_preflight(client: TestClient):
    r = client.options(
        "/tasks/",
        headers={"Origin": "http://127.0.0.1:5500", "Access-Control-Request-Method": "GET"},
    )
    assert r.headers["access-control-allow-origin"] == "http://127.0.0.1:5500"

Session 3 — PROJECT: Complete Todo API (users + auth + SQLite) on GitHub

Session flow

1. Homework review; all the tests from Session 2 should pass
2. Reading the project requirements together (list below)
3. The student writes it themselves (the teacher watches their screen and helps only if they get stuck)
4. Tests for the new features; the whole pytest run should pass
5. README, requirements.txt, .gitignore, push to GitHub
6. Student demo (they show their API in /docs) + Q&A + homework

Explanation for the teacher (read this first)

Today the student is the one sharing the screen, not you. Your job is to ask questions and help them if they stay stuck for a long time. Don't write the code for them — guide them ("Look at what the error says. Which line? Does this endpoint have CurrentUser?").

The foundation is already there (Sessions 1–2). The project adds four new features and "professional quality":

Project requirements (read them to the student):

1. Everything from Sessions 1–2 works (register, token, me, CRUD of owned tasks, CORS).
2. GET /tasks/stats → {"jumla": n, "zimekamilika": k, "zimebaki": n-k} (total, completed, remaining) for the logged-in user.
3. Search: GET /tasks/?q=neno filters by title.
4. POST /auth/change-password with body {"old_password": "...", "new_password": "..."} → 204; if the old password is wrong → 400.
5. SECRET_KEY from an environment variable (already in config.py) — the README should explain how to set it.
6. Tests: at least 12, all passing, including ones for features 2–4.
7. Repo todo-api on GitHub with README.md, requirements.txt, .gitignore (.venv/, __pycache__/, *.db, .env), pyproject.toml.

Setting an environment variable in PowerShell (for that terminal session only):

$env:SECRET_KEY = python -c "import secrets; print(secrets.token_hex(32))"
fastapi dev app/main.py

Marking criteria (rubric) — 20 marks:

Say this to the student

"Today you are the master craftsman. I am only the supervisor — I'll ask questions. If you get stuck, read the error out loud first; half the answers are inside the error itself."

"This project is the first you can put on your CV: 'I built a REST API with JWT authentication, a database, and tests.' That is a big deal — employers look for it."

"Remember the route order trap: /stats must be above /{task_id}."

Type this code live (share screen)

Today the student does the typing; use these pieces only if they are stuck (or to show at the end of the session).

Final structure:

todo_api/
├── .gitignore
├── README.md
├── pyproject.toml
├── requirements.txt
├── app/
│   ├── __init__.py
│   ├── config.py
│   ├── database.py
│   ├── deps.py
│   ├── main.py
│   ├── models.py
│   ├── security.py
│   └── routers/
│       ├── __init__.py
│       ├── auth.py
│       └── tasks.py
└── tests/
    ├── conftest.py
    ├── test_auth.py
    └── test_tasks.py

Search (q) in list_tasks — add a parameter and one line:

    q: str | None = None,
):
    query = select(Task).where(Task.owner_id == current_user.id)
    if done is not None:
        query = query.where(Task.done == done)
    if q:
        query = query.where(Task.title.contains(q))

stats — ABOVE @router.get("/{task_id}"):

@router.get("/stats")
def stats(session: SessionDep, current_user: CurrentUser) -> dict[str, int]:
    tasks = session.exec(select(Task).where(Task.owner_id == current_user.id)).all()
    zimekamilika = sum(1 for t in tasks if t.done)
    return {"jumla": len(tasks), "zimekamilika": zimekamilika, "zimebaki": len(tasks) - zimekamilika}

change-password — the model in app/models.py:

class PasswordChange(SQLModel):
    old_password: str
    new_password: str = Field(min_length=6, max_length=128)

and the endpoint in app/routers/auth.py (add PasswordChange to the models import):

@router.post("/change-password", status_code=status.HTTP_204_NO_CONTENT)
def change_password(data: PasswordChange, current_user: CurrentUser, session: SessionDep):
    if not verify_password(data.old_password, current_user.hashed_password):
        raise HTTPException(status_code=400, detail="Password ya zamani si sahihi")
    current_user.hashed_password = hash_password(data.new_password)
    session.add(current_user)
    session.commit()

Why does session.add(current_user) work? Because get_current_user got current_user using the same session (FastAPI calls get_session once per request).

.gitignore:

.venv/
__pycache__/
.pytest_cache/
*.db
.env

README.md (example):

# Todo API

REST API ya kusimamia tasks, iliyojengwa kwa FastAPI, SQLModel (SQLite), na JWT auth.

## Kuendesha
```powershell
python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt
$env:SECRET_KEY = python -c "import secrets; print(secrets.token_hex(32))"
fastapi dev app/main.py
```
Fungua http://127.0.0.1:8000/docs

## Endpoints
| Method | Path | Maelezo | Login? |
|--------|------|---------|--------|
| POST | /auth/register | Kujisajili | Hapana |
| POST | /auth/token | Kuingia (form data) | Hapana |
| GET | /auth/me | Taarifa zangu | Ndiyo |
| POST | /auth/change-password | Kubadilisha password | Ndiyo |
| GET/POST | /tasks/ | Orodha / kuunda | Ndiyo |
| GET | /tasks/stats | Takwimu | Ndiyo |
| GET/PATCH/DELETE | /tasks/{id} | Task moja | Ndiyo |

## Tests
```powershell
pytest -v
```

Git:

pip freeze > requirements.txt
pytest
git init
git add .
git status          # hakikisha .venv na todo.db HAZIMO
git commit -m "Todo API: auth ya JWT, tasks, tests"
git branch -M main
git remote add origin https://github.com/<jina-lako>/todo-api.git
git push -u origin main

Questions to ask the student

1. Show me: what path does a GET /tasks/5 request take through your code to get an answer? — Router tasks → read_task → OwnTask → get_own_task → (SessionDep, CurrentUser → get_current_user → oauth2_scheme → token → decode_access_token) → session.get → ownership → response through TaskPublic.
2. If you change the SECRET_KEY while the server is running and people are logged in, what happens? — All old tokens become invalid (401); everyone must log in again.
3. Why does change-password need the old password when the user already has a token? — If the token was stolen, the thief must not be able to change the password and lock the owner out of the account.
4. Why is .env in .gitignore? — It holds secrets (such as the SECRET_KEY).
5. What does your "second user doesn't see the task" test protect? — Authorization/ownership — a mistake here would leak people's data.

Common mistakes

In-class exercise

The project itself is the exercise. For tests of the new features:

Exercise 3.1 — A stats test: create 2 tasks (one done: true), confirm {"jumla": 2, "zimekamilika": 1, "zimebaki": 1}.

Exercise 3.2 — A change-password test: wrong old password → 400; correct → 204; login with the old one → 401; with the new one → 200.

Exercise 3.3 — A q test: create "Kununua sukari" (buying sugar) and "Kusoma" (studying), GET /tasks/?q=sukari should return one.

# kwenye tests/test_tasks.py
def test_stats(client, auth_headers):
    client.post("/tasks/", json={"title": "a"}, headers=auth_headers)
    client.post("/tasks/", json={"title": "b", "done": True}, headers=auth_headers)
    r = client.get("/tasks/stats", headers=auth_headers)
    assert r.json() == {"jumla": 2, "zimekamilika": 1, "zimebaki": 1}


def test_search(client, auth_headers):
    client.post("/tasks/", json={"title": "Kununua sukari"}, headers=auth_headers)
    client.post("/tasks/", json={"title": "Kusoma"}, headers=auth_headers)
    r = client.get("/tasks/", params={"q": "sukari"}, headers=auth_headers)
    assert [t["title"] for t in r.json()] == ["Kununua sukari"]


# kwenye tests/test_auth.py
def test_change_password(client, auth_headers):
    r = client.post("/auth/change-password",
                    json={"old_password": "kosa", "new_password": "mpya1234"},
                    headers=auth_headers)
    assert r.status_code == 400

    r = client.post("/auth/change-password",
                    json={"old_password": "siri123", "new_password": "mpya1234"},
                    headers=auth_headers)
    assert r.status_code == 204

    assert client.post("/auth/token", data={"username": "amina", "password": "siri123"}).status_code == 401
    assert client.post("/auth/token", data={"username": "amina", "password": "mpya1234"}).status_code == 200

Homework

To submit: next session.

1. Priority and due date: add priority: int = Field(default=3, ge=1, le=5) and due_date: date | None = None to TaskBase (and TaskUpdate). Add a query parameter panga (sort) to GET /tasks/ that accepts "id", "priority", "due_date" (anything else → 400).
2. GET /tasks/overdue — the user's tasks whose due_date has passed (earlier than today) and done == False.
3. DELETE /auth/me — the user deletes their account along with all their tasks → 204.
4. 3 new tests for these features.
5. Question (write it in the README under a "Nilichojifunza" (What I learned) section): Explain in 3–5 sentences the token's journey: from login to GET /tasks/.

from datetime import date, datetime, timezone

class TaskBase(SQLModel):
    title: str = Field(min_length=1, max_length=100)
    description: str | None = None
    done: bool = False
    priority: int = Field(default=3, ge=1, le=5)
    due_date: date | None = None


class TaskUpdate(SQLModel):
    title: str | None = Field(default=None, min_length=1, max_length=100)
    description: str | None = None
    done: bool | None = None
    priority: int | None = Field(default=None, ge=1, le=5)
    due_date: date | None = None

from datetime import date

SORT_COLUMNS = {"id": Task.id, "priority": Task.priority, "due_date": Task.due_date}


@router.get("/", response_model=list[TaskPublic])
def list_tasks(
    session: SessionDep,
    current_user: CurrentUser,
    done: bool | None = None,
    q: str | None = None,
    panga: str = "id",
    offset: int = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
    if panga not in SORT_COLUMNS:
        raise HTTPException(status_code=400, detail=f"panga iwe moja ya {list(SORT_COLUMNS)}")
    query = select(Task).where(Task.owner_id == current_user.id)
    if done is not None:
        query = query.where(Task.done == done)
    if q:
        query = query.where(Task.title.contains(q))
    query = query.order_by(SORT_COLUMNS[panga]).offset(offset).limit(limit)
    return session.exec(query).all()


# juu ya /{task_id}
@router.get("/overdue", response_model=list[TaskPublic])
def overdue(session: SessionDep, current_user: CurrentUser):
    query = select(Task).where(
        Task.owner_id == current_user.id,
        Task.done == False,  # noqa: E712
        Task.due_date != None,  # noqa: E711
        Task.due_date < date.today(),
    )
    return session.exec(query).all()

from app.models import Task  # ongeza kwenye imports


@router.delete("/me", status_code=status.HTTP_204_NO_CONTENT)
def delete_me(current_user: CurrentUser, session: SessionDep):
    tasks = session.exec(select(Task).where(Task.owner_id == current_user.id)).all()
    for t in tasks:
        session.delete(t)
    session.delete(current_user)
    session.commit()

from datetime import date, timedelta


def test_panga_mbaya(client, auth_headers):
    assert client.get("/tasks/?panga=rangi", headers=auth_headers).status_code == 400


def test_overdue(client, auth_headers):
    jana = (date.today() - timedelta(days=1)).isoformat()
    kesho = (date.today() + timedelta(days=1)).isoformat()
    client.post("/tasks/", json={"title": "Imechelewa", "due_date": jana}, headers=auth_headers)
    client.post("/tasks/", json={"title": "Bado", "due_date": kesho}, headers=auth_headers)
    client.post("/tasks/", json={"title": "Imeisha", "due_date": jana, "done": True}, headers=auth_headers)
    r = client.get("/tasks/overdue", headers=auth_headers)
    assert [t["title"] for t in r.json()] == ["Imechelewa"]


def test_delete_me(client, auth_headers):
    client.post("/tasks/", json={"title": "x"}, headers=auth_headers)
    assert client.delete("/auth/me", headers=auth_headers).status_code == 204
    # token bado ni "halali" kisaini, lakini user hayupo tena -> 401
    assert client.get("/auth/me", headers=auth_headers).status_code == 401

If you are asked a question you don't know

Don't make up an answer about security — a security mistake is costly. Say: "Good question; security is a broad area, let's check the official docs". Search: "fastapi refresh token", "fastapi oauth2 scopes", "OWASP password storage", "pytest fixture scope". Official sources: FastAPI Security https://fastapi.tiangolo.com/tutorial/security/ · FastAPI Testing https://fastapi.tiangolo.com/tutorial/testing/ · PyJWT https://pyjwt.readthedocs.io/ · pwdlib https://frankie567.github.io/pwdlib/ · pytest https://docs.pytest.org/ · MDN CORS https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS

End-of-week checklist

• Explains authentication vs authorization, 401 vs 403
• Explains hashing + salt and uses pwdlib (hash, verify)
• Explains the 3 parts of a JWT and that the payload isn't secret
• Knows /auth/token receives form data and returns access_token + token_type
• Uses Authorization: Bearer <token> with requests and in /docs (Authorize)
• Protects endpoints with CurrentUser and enforces ownership (owner_id)
• Explains origin and CORS, and sets up CORSMiddleware
• Writes tests with TestClient + pytest fixtures + dependency_overrides
• pytest passes everything (≥ 12)
• SECRET_KEY isn't hardcoded in the repo as a real secret; it is read from the environment
• The todo-api repo is on GitHub with a correct README, requirements.txt, .gitignore
• The student can demo their API and explain each part