Learn / Python and FastAPI Backend Engineering
Observability and Error Contracts
Design a FastAPI failure surface with stable ProblemDetails-style responses, request IDs, structured logs, bounded metrics, and a diagnostic playbook operators can actually follow.
Course: Python and FastAPI Backend Engineering. Level: Intermediate. Topic: Backend craft.
Stage: advanced - Advanced failure contracts - Observable errors and production diagnostics. Make expected failures predictable for clients and surprising failures diagnosable for operators.
Outcomes
- Map validation, auth, conflict, rate-limit, dependency, and unexpected failures into stable client-safe error contracts.
- Attach one request ID to the response, logs, and metrics so support can trace a failure without exposing internals.
- Create low-cardinality observability signals and a diagnostic playbook for inconsistent production errors.
- Raise exceptions for exceptional states and return values for expected branches.
- Create useful error messages without leaking secrets.
- Log structured context at the boundary where errors are handled.
- Apply timeouts and retries carefully around external I/O.
Concepts
- ProblemDetails error contract
- request ID middleware
- structured logging
- low-cardinality metric
- diagnostic playbook
- Errors and Reliability
- Exceptions are control boundaries
- Logging for diagnosis
- Python + FastAPI foundations
- Guided practice
Concept flow
Connect the client-safe problem response to the private log and bounded metric operators use to diagnose the same failure.
- Client request
- Request ID middleware
- Exception handler
- Problem response
- Structured log
- Metrics counter
- Diagnostic playbook
Session flow
- Write the failure matrix before handlers (concept, 12 min) — Map common failure categories to status, public code, safe detail, log level, and metric labels.
- A failure matrix keeps API errors consistent across routes.
- Expected failures and unexpected failures need different logging and client promises.
- Propagate request IDs everywhere useful (walkthrough, 16 min) — Accept or generate a request id, store it on request.state, echo it in headers, and attach it to logs.
- Correlation is only real when every surface uses the same id.
- Validate or bound incoming request ids before trusting them in logs.
- Separate public problem responses from private diagnostics (walkthrough, 18 min) — Implement ApiError and unexpected exception handlers that share a response shape but differ in log detail.
- ProblemDetails-style responses give clients predictable fields.
- Logger.exception belongs in the unexpected path, not the public body.
- Turn signals into a diagnostic playbook (exercise, 10 min) — Write the operator path from request_id to log query, metric panel, suspected owner, and customer-safe update.
- Observability is incomplete until someone can use it under pressure.
- Runbooks should name the first query and the rollback or escalation signal.
- Exceptions, Logging, and Failure Boundaries: Exceptions are control boundaries (concept, 60 min) — An exception should carry enough information for the caller to handle or report the failure. Catch exceptions at meaningful boundaries, such as an HTTP route, worker job, or command-line entry point.
- Retained source example: Domain-specific exception
class InsufficientStockError(Exception):
pass
def reserve(stock: int, requested: int) -> int:
if requested > stock:
raise InsufficientStockError('not enough stock')
return stock - requested
print(reserve(5, 2))
Expected output: 3
- Exceptions, Logging, and Failure Boundaries: Logging for diagnosis (walkthrough, 60 min) — Logs should answer what happened, where it happened, and what context is safe to record. Do not log passwords, tokens, full payment details, or personal data unless policy explicitly allows it.
- Retained source example: Boundary logging
import logging
logger = logging.getLogger(__name__)
def handle_order(order_id: str) -> None:
try:
raise TimeoutError('payment provider timed out')
except TimeoutError:
logger.exception('order processing failed', extra={'order_id': order_id})
- Exceptions, Logging, and Failure Boundaries: practices and mistakes (review, 5 min) — Retained guidance from the legacy PTesting lesson.
- Practice: Catch specific exceptions and add context before re-raising when useful.
- Practice: Use timeouts for network and database operations.
- Practice: Return safe, user-oriented messages while logging diagnostic details internally.
- Avoid: Using exceptions for normal optional results.
- Avoid: Logging secrets or whole request bodies.
- Avoid: Retrying non-idempotent operations without safeguards.
- Exceptions, Logging, and Failure Boundaries: references (review, 2 min) — Original references retained from the legacy library.
- Python tutorial: https://docs.python.org/3/tutorial/
- Python standard library: https://docs.python.org/3/library/
- logging module: https://docs.python.org/3/library/logging.html
- FastAPI error handling: https://fastapi.tiangolo.com/tutorial/handling-errors/
Code example
Python in app/observability/errors.py.
from dataclasses import dataclass
from http import HTTPStatus
from time import perf_counter
from uuid import uuid4
import structlog
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
logger = structlog.get_logger()
@dataclass(frozen=True)
class ApiError(Exception):
status_code: int
code: str
title: str
detail: str
public_meta: dict[str, str] | None = None
def request_id_from(request: Request) -> str:
header_value = request.headers.get("x-request-id")
if header_value and len(header_value) <= 80:
return header_value
return str(uuid4())
def problem_response(
request: Request,
*,
status_code: int,
code: str,
title: str,
detail: str,
request_id: str,
meta: dict[str, str] | None = None,
) -> JSONResponse:
body = {
"type": f"https://api.example.com/problems/{code}",
"title": title,
"status": status_code,
"detail": detail,
"code": code,
"request_id": request_id,
"instance": str(request.url.path),
}
if meta:
body["meta"] = meta
return JSONResponse(
status_code=status_code,
content=body,
headers={"x-request-id": request_id},
media_type="application/problem+json",
)
def create_app(metrics: Metrics) -> FastAPI:
app = FastAPI()
@app.middleware("http")
async def request_context(request: Request, call_next):
request_id = request_id_from(request)
request.state.request_id = request_id
started_at = perf_counter()
response = await call_next(request)
elapsed_ms = round((perf_counter() - started_at) * 1000, 2)
response.headers["x-request-id"] = request_id
logger.info(
"http_request",
request_id=request_id,
route=request.scope.get("route").path if request.scope.get("route") else request.url.path,
status_code=response.status_code,
elapsed_ms=elapsed_ms,
)
return response
@app.exception_handler(ApiError)
async def api_error_handler(request: Request, exc: ApiError):
request_id = request.state.request_id
status_family = f"{exc.status_code // 100}xx"
logger.warning(
"api_error",
request_id=request_id,
code=exc.code,
status_code=exc.status_code,
route=request.scope.get("route").path if request.scope.get("route") else request.url.path,
)
metrics.increment(
"api_errors_total",
labels={"code": exc.code, "status_family": status_family},
)
return problem_response(
request,
status_code=exc.status_code,
code=exc.code,
title=exc.title,
detail=exc.detail,
request_id=request_id,
meta=exc.public_meta,
)
@app.exception_handler(Exception)
async def unexpected_error_handler(request: Request, exc: Exception):
request_id = request.state.request_id
logger.exception("unexpected_error", request_id=request_id)
metrics.increment(
"api_errors_total",
labels={"code": "internal_error", "status_family": "5xx"},
)
return problem_response(
request,
status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
code="internal_error",
title="Internal server error",
detail="Something went wrong. Share request_id with support.",
request_id=request_id,
)
return app
Walkthrough examples
- Failure contract and diagnostics pack — A production API sometimes returns inconsistent errors. Support needs one request id, clients need stable fields, and operators need enough private context to diagnose safely.
- File: app/observability/errors.py
- File: app/observability/request_context.py
- File: app/observability/metrics.py
- File: tests/test_error_contracts.py
- File: docs/ops/error-diagnostics.md
- Create an ApiError class and problem_response helper with stable fields and application/problem+json.
- Add request ID middleware that echoes x-request-id and stores the same value on request.state.
- Write exception handlers for expected ApiError and unexpected Exception paths.
- Add route tests for conflict, dependency failure, and unexpected exception without leaking stack traces.
- Document the log query, metric panel, owner escalation, rollback signal, and customer-safe support wording.
- Domain-specific exception — Additional example retained from the legacy lesson.
- Retained source code:
class InsufficientStockError(Exception):
pass
def reserve(stock: int, requested: int) -> int:
if requested > stock:
raise InsufficientStockError('not enough stock')
return stock - requested
print(reserve(5, 2))
- Expected output: 3
- Compare the example with the canonical PTLearn implementation.
- Boundary logging — Additional example retained from the legacy lesson.
- Retained source code:
import logging
logger = logging.getLogger(__name__)
def handle_order(order_id: str) -> None:
try:
raise TimeoutError('payment provider timed out')
except TimeoutError:
logger.exception('order processing failed', extra={'order_id': order_id})
- Explain the expected behavior.
- Compare the example with the canonical PTLearn implementation.
- Validate early — Additional example retained from the legacy lesson.
- Retained source code:
def page_size(value: int) -> int:
if not 1 <= value <= 100:
raise ValueError('page size must be 1..100')
return value
- Explain the expected behavior.
- Compare the example with the canonical PTLearn implementation.
Practice
- Write a failure matrix for validation, auth, not-found, conflict, rate-limit, dependency, and unexpected errors.
- Return one stable application/problem+json body for an ApiError and assert its code, status, request_id, and content type.
- Log the same request_id with sanitized context and no secrets, tokens, raw SQL, or stack traces in the response body.
- Track error metrics with bounded labels such as error code and status family, not request id or raw exception text.
- Write a five-step diagnostic playbook that starts from a public request_id and ends with a safe owner action.
- Choose the handling boundary: Where should an API service usually translate an internal domain error into an HTTP status code?
- Hint: Think about the web boundary.
- Reference solution: At the route or API exception-handler boundary.
- Accepted answers: At the route or API exception-handler boundary. | route boundary | exception handler
Checklist
- Every public error has a stable code, status, title, detail, request_id, and content type.
- Unexpected exceptions return a generic client-safe body while logs keep private diagnostic context.
- Request IDs are accepted when safe, generated when missing, and echoed in response headers.
- Structured logs include request_id, route template, status code, and error code.
- Metrics use low-cardinality labels and avoid user ids, emails, raw paths, request ids, and exception messages.
- The diagnostic playbook names first query, owner escalation, rollback signal, and customer-safe support wording.
Quiz prompts
- Why should a client-facing 500 response avoid the Python stack trace? — Public responses should help clients and support without leaking internals; protected logs carry the debugging detail.
- Support receives a user screenshot with request_id. What must be true for that id to be useful? — Correlation only works when the same safe value is present in the user-visible response and the operator-visible log.
- Which metric label set is safest for api_errors_total? — Bounded labels keep metrics useful and affordable; high-cardinality or sensitive labels belong nowhere near a metric stream.
- What is the best first test for an ApiError handler? — The public error contract is an HTTP boundary, so test the response shape through the app.
- A teammate wants to include tenant_id in every public error body. What is the review question? — Public metadata should be intentionally safe and useful; correlation and diagnostics can stay in logs.
- What is a danger of except Exception: pass? — Swallowing broad exceptions removes the signal needed to debug and recover.
Flashcards
- What fields make a useful public problem response? Stable type/code, title, status, safe detail, request_id, and optionally client-safe metadata.
- What makes a request ID useful? The same bounded safe value appears in the response header/body and the structured logs for that request.
- What belongs in logs but not public errors? Exception details, stack traces, dependency context, and sanitized internal state needed by operators.
- Why avoid request_id as a metric label? It creates one label value per request, exploding cardinality and making metrics expensive and hard to query.
- What should an error diagnostic playbook start with? A public request_id or error code, then the exact log query, metric check, owner path, and safe user update.
- How should expected domain errors differ from unexpected exceptions? Expected errors use specific public codes and warning logs; unexpected exceptions use generic public details and private exception logs.
Labs
- Ship an observable error surface — Replace ad hoc FastAPI errors with a stable problem response contract, correlated logs, bounded metrics, and a diagnostic playbook.
- Create ApiError, problem_response, and exception handlers for expected and unexpected failures.
- Add request ID middleware that accepts a safe x-request-id or generates one.
- Write tests for a known conflict, a dependency outage, an unexpected exception, and request_id header/body consistency.
- Add a metrics counter with bounded labels for error code and status family.
- Write docs/ops/error-diagnostics.md with the first log query, metric panel, escalation owner, rollback signal, and customer-safe wording.
- All tested error responses use application/problem+json and stable public fields.
- The same request_id appears in response headers, response body, and captured logs.
- Unexpected exceptions never expose stack traces or dependency internals to clients.
- Metrics labels exclude request ids, emails, raw paths, exception messages, and secrets.
- The playbook lets another engineer start from a user report and find the matching log.
Challenge
- Review-ready FastAPI failure surface (Stretch) — Turn scattered HTTPException usage into a documented failure surface with a response matrix, stable problem responses, correlated logs, bounded metrics, and a diagnostic playbook.
- Failure matrix covers validation, auth, not-found, conflict, rate-limit, dependency, and unexpected errors.
- Tests prove response shape, request_id propagation, content type, and no internal leak for expected and unexpected failures.
- Structured logs and metrics carry enough bounded context to diagnose without exposing secrets.
- The runbook starts from a request_id and names query, dashboard, owner, rollback signal, and support response.
Canonical lesson URL