Request Logging Middleware

Add middleware that logs every request with its method, path, status, and duration

💻

Writing code and entering commands is only available on desktop. Open this page on a larger screen to complete this chapter.

Why log requests

When your API is running, you have no visibility into what is happening unless you add logging. If a user reports a slow response or an error, you need to know which endpoint was hit, what status code was returned, and how long it took.

A logging middleware prints this information for every request automatically. You do not need to add logging code to each endpoint individually.

Measuring duration

To measure how long a request takes, record the time before calling the endpoint and subtract it from the time after. Python's time.time() returns the current time in seconds. Multiplying the difference by 1000 converts it to milliseconds.

start = time.time()
# ... do work ...
duration = (time.time() - start) * 1000

The middleware pattern

Your middleware will:

Record the start time
Call call_next(request) to run the endpoint
Calculate the duration
Print the method, path, status code, and duration
Return the response

Instructions

Add a request logging middleware to your API.

Add import time to the top of the file, after the existing imports.
Add the @app.middleware("http") decorator after the load_expenses() call and before the first endpoint.
Define an async function called log_requests that takes request and call_next as parameters.
Inside the function, record the start time with start = time.time().
Call response = await call_next(request) — this passes the request to the next handler (your endpoint) and waits for the response.
Calculate the duration with duration = (time.time() - start) * 1000.
Print the log line: f"{request.method} {request.url.path} {response.status_code} {duration:.1f}ms".
Return response.

← Previous Chapter Next Chapter →

import json
from datetime import datetime
from pathlib import Path
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, field_validator
from typing import Literal, Optional

# Step 1: Add import time

app = FastAPI()
DATA_FILE = Path("expenses.json")

class Expense(BaseModel):
    description: str = Field(min_length=1)
    amount: float = Field(gt=0)
    category: Literal["food", "transport", "entertainment", "utilities", "other"]
    date: Optional[str] = None

    @field_validator("date")
    @classmethod
    def validate_date_format(cls, v):
        if v is not None:
            try:
                datetime.strptime(v, "%Y-%m-%d")
            except ValueError:
                raise ValueError("Date must be in YYYY-MM-DD format")
        return v

class ExpenseUpdate(BaseModel):
    description: Optional[str] = Field(default=None, min_length=1)
    amount: Optional[float] = Field(default=None, gt=0)
    category: Optional[Literal["food", "transport", "entertainment", "utilities", "other"]] = None

class ExpenseOut(BaseModel):
    id: int
    description: str
    amount: float
    category: str
    date: Optional[str] = None

class ErrorResponse(BaseModel):
    detail: str

def save_expenses():
    data = {"counter": counter, "expenses": expenses}
    DATA_FILE.write_text(json.dumps(data, indent=2))

def load_expenses():
    global expenses, counter
    if DATA_FILE.exists():
        data = json.loads(DATA_FILE.read_text())
        expenses = {int(k): v for k, v in data["expenses"].items()}
        counter = data["counter"]

expenses = {}
counter = 0
load_expenses()

# Step 2: Add @app.middleware("http")
# Step 3: Define async def log_requests(request, call_next)
# Steps 4-8: start time, await call_next, duration, print, return
#   Record start time
#   Call response = await call_next(request)
#   Calculate duration in ms
#   Print method, path, status_code, duration
#   Return response

@app.post("/expenses", status_code=201, response_model=ExpenseOut)
def create_expense(expense: Expense):
    global counter
    counter += 1
    expenses[counter] = {"id": counter, **expense.model_dump()}
    save_expenses()
    return expenses[counter]

@app.get("/expenses", response_model=list[ExpenseOut])
def list_expenses(category: str = None):
    if category:
        return [e for e in expenses.values() if e["category"] == category]
    return list(expenses.values())

@app.get("/expenses/{expense_id}", response_model=ExpenseOut, responses={404: {"model": ErrorResponse}})
def get_expense(expense_id: int):
    if expense_id not in expenses:
        raise HTTPException(status_code=404, detail="Expense not found")
    return expenses[expense_id]

@app.delete("/expenses/{expense_id}", response_model=ExpenseOut, responses={404: {"model": ErrorResponse}})
def delete_expense(expense_id: int):
    if expense_id not in expenses:
        raise HTTPException(status_code=404, detail="Expense not found")
    save_expenses()
    return expenses.pop(expense_id)

@app.patch("/expenses/{expense_id}", response_model=ExpenseOut, responses={404: {"model": ErrorResponse}})
def update_expense(expense_id: int, updates: ExpenseUpdate):
    if expense_id not in expenses:
        raise HTTPException(status_code=404, detail="Expense not found")
    expenses[expense_id].update(updates.model_dump(exclude_unset=True))
    save_expenses()
    return expenses[expense_id]

@app.get("/summary")
def spending_summary():
    summary = {}
    for expense in expenses.values():
        cat = expense["category"]
        summary[cat] = summary.get(cat, 0) + expense["amount"]
    return summary

Request Logging Middleware

Why log requests

Measuring duration

The middleware pattern

Instructions

Interactive Code Editor