PYTHON FASTAPI · LESSON 7 OF 8

Test success, errors and access control

Run repeatable checks without touching your development database.

pytest
TestClient
fixtures
isolation

Your runnable checkpoint

Extract into a new practice folder. Each download includes complete source, requirements.txt and README.md. Stop any earlier server before running this example.

Download lesson source ↓

Create and activate a virtual environment as shown in lesson 1. Install with python -m pip install -r requirements.txt, then run fastapi dev main.py. Lessons 6–8 require TASK_API_KEY as explained in lesson 6.

1. Understand the concepts

Manual requests help you explore; automated tests tell you whether a change broke behavior. Save test_main.py beside main.py and run python -m pytest -q from that directory. The server does not need to be running.

TestClient calls the ASGI application in-process. The with block triggers lifespan, creating the test table. tmp_path gives each test its own temporary database. monkeypatch replaces configuration for a test and restores it afterwards.

The lifecycle test creates its own task, captures its id, updates it and deletes it. It checks externally visible behavior rather than relying on a fixed database id. Validation and security tests exercise rejected requests as well as successful ones.

The test-only key is deliberately not a real credential. Do not point test configuration at production data. A test that merely receives 200 does not prove that authorization or persistence works; assert the body and error behavior too.

2. Build and run

These are complete files, not fragments. Save them beside each other in the extracted project folder. Open http://127.0.0.1:8000/docs to send requests.

main.py

import os
import sqlite3
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Query, Response
from fastapi.security import APIKeyHeader
import secrets
from pydantic import BaseModel, ConfigDict, Field

DB_PATH = os.getenv("TASK_DB", "tasks.db")

def get_db():
    connection = sqlite3.connect(DB_PATH, check_same_thread=False)
    connection.row_factory = sqlite3.Row
    try:
        yield connection
    finally:
        connection.close()

@asynccontextmanager
async def lifespan(app: FastAPI):
    connection = sqlite3.connect(DB_PATH)
    try:
        connection.execute("CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, title TEXT NOT NULL, done INTEGER NOT NULL DEFAULT 0)")
        connection.commit()
    finally:
        connection.close()
    yield

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

def require_api_key(key: Annotated[str | None, Depends(api_key_header)]):
    expected = os.getenv("TASK_API_KEY")
    if not expected:
        raise HTTPException(status_code=503, detail="API key is not configured")
    if not key or not secrets.compare_digest(key, expected):
        raise HTTPException(status_code=401, detail="Invalid API key")

app = FastAPI(title="EduPoly Task API", lifespan=lifespan)
Database = Annotated[sqlite3.Connection, Depends(get_db)]

class TaskCreate(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
    title: str = Field(min_length=1, max_length=120)
    done: bool = False

class TaskRead(TaskCreate):
    id: int

def find_task(db, task_id):
    row = db.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return dict(row)

@app.get("/health")
def health():
    return {"status": "ok"}

@app.post("/tasks", response_model=TaskRead, status_code=201, dependencies=[Depends(require_api_key)])
def create_task(task: TaskCreate, db: Database):
    cursor = db.execute("INSERT INTO tasks (title, done) VALUES (?, ?)", (task.title, task.done))
    db.commit()
    return find_task(db, cursor.lastrowid)

@app.get("/tasks", response_model=list[TaskRead], dependencies=[Depends(require_api_key)])
def list_tasks(db: Database, limit: Annotated[int, Query(ge=1, le=100)] = 20):
    return [dict(row) for row in db.execute("SELECT * FROM tasks ORDER BY id LIMIT ?", (limit,))]

@app.get("/tasks/{task_id}", response_model=TaskRead, dependencies=[Depends(require_api_key)])
def read_task(task_id: int, db: Database):
    return find_task(db, task_id)

@app.put("/tasks/{task_id}", response_model=TaskRead, dependencies=[Depends(require_api_key)])
def replace_task(task_id: int, task: TaskCreate, db: Database):
    find_task(db, task_id)
    db.execute("UPDATE tasks SET title = ?, done = ? WHERE id = ?", (task.title, task.done, task_id))
    db.commit()
    return find_task(db, task_id)

@app.delete("/tasks/{task_id}", status_code=204, dependencies=[Depends(require_api_key)])
def delete_task(task_id: int, db: Database):
    find_task(db, task_id)
    db.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
    db.commit()
    return Response(status_code=204)

requirements.txt

fastapi[standard]>=0.115,<1
pydantic>=2,<3
httpx>=0.27,<1
pytest>=8,<10

test_main.py

import pytest
from fastapi.testclient import TestClient
import main

@pytest.fixture
def client(tmp_path, monkeypatch):
    monkeypatch.setattr(main, "DB_PATH", str(tmp_path / "test.db"))
    monkeypatch.setenv("TASK_API_KEY", "test-only-key")
    with TestClient(main.app) as session:
        session.headers["X-API-Key"] = "test-only-key"
        yield session

def test_task_lifecycle(client):
    created = client.post("/tasks", json={"title": "Learn FastAPI"})
    assert created.status_code == 201
    task_id = created.json()["id"]
    assert created.json()["done"] is False
    assert client.get(f"/tasks/{task_id}").status_code == 200
    updated = client.put(f"/tasks/{task_id}", json={"title": "Finished", "done": True})
    assert updated.json()["done"] is True
    assert len(client.get("/tasks").json()) == 1
    deleted = client.delete(f"/tasks/{task_id}")
    assert deleted.status_code == 204
    assert deleted.content == b""
    assert client.get(f"/tasks/{task_id}").status_code == 404

def test_validation(client):
    assert client.post("/tasks", json={"title": "  "}).status_code == 422
    assert client.post("/tasks", json={"title": "Valid", "typo": 1}).status_code == 422
    assert client.get("/tasks?limit=101").status_code == 422

def test_security(client):
    client.headers.pop("X-API-Key")
    assert client.get("/tasks").status_code == 401
    assert client.get("/health").status_code == 200
    client.headers["X-API-Key"] = "wrong"
    assert client.post("/tasks", json={"title": "Blocked"}).status_code == 401

def test_missing_configuration(client, monkeypatch):
    monkeypatch.delenv("TASK_API_KEY")
    assert client.get("/tasks").status_code == 503

3. Check the result

python -m pytest -q reports four passing tests.
Your existing tasks.db is unchanged.
Change the expected create status to 200 and observe the test fail; restore 201.

4. Practice independently

Test DELETE of an id that does not exist.

Show the explanation

Call client.delete("/tasks/999999") on the isolated database and assert status_code == 404.

When something goes wrong

For import errors, confirm the virtual environment is active and install requirements.txt using that environment’s Python. For 422, read the response detail and compare your input to the model. For 404, check the method, path and resource id. For 401 in later lessons, use Authorize in /docs with the same key configured on the server.