PYTHON FASTAPI · LESSON 6 OF 8

Protect the API with a dependency

Require a secret header for every task operation.

APIKeyHeader
environment
401
authorization

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

A shared API key is suitable for a small server-to-server demo. It is not individual user login: every key holder can access every task. Private per-user tasks need identity, ownership checks and an authentication design beyond this tutorial.

Generate a secret with python -c "import secrets; print(secrets.token_urlsafe(32))". On macOS/Linux set export TASK_API_KEY="your-generated-value"; in Windows Command Prompt use set TASK_API_KEY=your-generated-value. Set it in the terminal that starts FastAPI. Never commit the value or embed it in frontend JavaScript.

APIKeyHeader reads X-API-Key and adds an Authorize control in /docs. require_api_key runs before each task route. secrets.compare_digest compares secrets; missing or incorrect keys return 401. A missing server configuration returns 503 instead of accidentally allowing access.

Authorize in /docs, then exercise the same CRUD requests. /health stays public and reveals no task data. Use HTTPS when traffic leaves your machine. CORS is a browser policy and does not replace authentication; do not expose a shared key in a browser client.

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

3. Check the result

Without X-API-Key, GET /tasks returns 401 when the server key is configured.
Use Authorize in /docs: a valid key allows CRUD.
GET /health remains public.

4. Practice independently

Explain why an API key in a React application is not a secret.

Show the explanation

The browser downloads JavaScript and sends requests visible to the user. Keep service keys on a trusted server; use user authentication and ownership checks for a browser-facing task app.

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.