FastAPI Backend Development Bootcamp · 课时

结构化 JSON 日志与关联 ID

输出带请求作用域关联 ID 的结构化日志，使其能够跨越异步边界和服务持续传递。

第 1 / 4 课13 个步骤

结构化 JSON 日志与关联 ID 是 CoddyKit 上的免费 FastAPI Backend Development Bootcamp 课时。这是第 1 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 FastAPI Backend Development Bootcamp 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

本课时的部分内容尚未翻译，以英文显示。

Why Structured Logs

In production, logs are data, not prose. A line like User 42 failed login from 10.0.0.3 reads fine to a human but is painful for machines: you cannot reliably filter, aggregate, or alert on it.

Structured logging emits each event as a JSON object with stable, queryable fields:

timestamp, level, message
request_id / correlation_id
context such as user_id, path, status_code, duration_ms

Log aggregators (Loki, Elasticsearch, Datadog) then index those fields so you can run queries like level=ERROR AND path=/checkout.

A JSON Log in One Line

The simplest structured log is just a dictionary serialized to JSON on one line. One JSON object per line is the JSON Lines (NDJSON) format that virtually every log shipper understands.

This standalone example shows the shape we are aiming for. Notice the fields are flat and named consistently.

import json
import time

def log(level, message, **fields):
    record = {
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "level": level,
        "message": message,
        **fields,
    }
    print(json.dumps(record))

log("INFO", "request completed", path="/checkout", status_code=200, duration_ms=42)
log("ERROR", "db timeout", path="/orders", correlation_id="abc-123")

A Custom JSON Formatter

Rolling your own print(json.dumps(...)) bypasses Python's logging module, losing levels, handlers, and library logs. Instead, plug a JSON formatter into the standard logging stack.

A formatter's job is to turn a LogRecord into a string. Here we return JSON. record.__dict__ carries any extra={...} fields you pass at the call site.

import json
import logging

class JsonFormatter(logging.Formatter):
    def format(self, record):
        payload = {
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload)

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])

logging.getLogger("app").info("service started", extra={"port": 8000})

The Correlation ID Problem

A single user request often fans out: API handler -> service layer -> database call -> outbound HTTP call to another service. If each log line is anonymous, you cannot stitch them back into one story.

A correlation ID (a.k.a. request ID or trace ID) is a unique value generated once per inbound request and attached to every log line produced while handling it. Then correlation_id=abc-123 retrieves the full timeline across functions and even across services.

The challenge: how do you make that ID available deep in the call stack without threading it through every function argument?

ContextVar: Request-Scoped State

The clean answer is contextvars.ContextVar. Unlike a global variable, a ContextVar holds a value that is isolated per logical execution context and, crucially, propagates correctly across async awaits.

Each concurrent request runs in its own context, so setting the correlation ID in one request never leaks into another, even when many run interleaved on the same event loop.

import asyncio
from contextvars import ContextVar

correlation_id: ContextVar[str] = ContextVar("correlation_id", default="-")

async def handle(name, cid):
    correlation_id.set(cid)
    await asyncio.sleep(0.01)
    # value survives the await and stays isolated per task
    print(name, "->", correlation_id.get())

async def main():
    await asyncio.gather(
        handle("req-A", "aaa"),
        handle("req-B", "bbb"),
    )

asyncio.run(main())

Injecting the ID via a Log Filter

To get the correlation ID onto every log line automatically, attach a logging.Filter that reads the ContextVar and copies it onto the record. A filter runs for every record passing through the handler, so no call site has to remember to pass the ID.

The formatter then reads record.correlation_id like any other field.

import json
import logging
from contextvars import ContextVar

correlation_id: ContextVar[str] = ContextVar("correlation_id", default="-")

class CorrelationFilter(logging.Filter):
    def filter(self, record):
        record.correlation_id = correlation_id.get()
        return True

class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "level": record.levelname,
            "message": record.getMessage(),
            "correlation_id": getattr(record, "correlation_id", "-"),
        })

h = logging.StreamHandler()
h.addFilter(CorrelationFilter())
h.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[h])

correlation_id.set("abc-123")
logging.getLogger("app").info("order placed")

FastAPI Middleware to Set the ID

In FastAPI, the right place to establish the correlation ID is an HTTP middleware, which wraps every request. The pattern:

Read an incoming X-Request-ID / X-Correlation-ID header if a caller (gateway, upstream service) already set one.
Otherwise generate a fresh UUID.
Store it in the ContextVar so all downstream logs pick it up.
Echo it back in the response header so clients can report it in bug reports.

This is framework code that needs a running server, so it is illustrative rather than runnable.

import uuid
from fastapi import FastAPI, Request
from contextvars import ContextVar

correlation_id: ContextVar[str] = ContextVar("correlation_id", default="-")
app = FastAPI()

@app.middleware("http")
async def correlation_middleware(request: Request, call_next):
    cid = request.headers.get("X-Request-ID") or str(uuid.uuid4())
    token = correlation_id.set(cid)
    try:
        response = await call_next(request)
    finally:
        correlation_id.reset(token)
    response.headers["X-Request-ID"] = cid
    return response

Why reset() with a Token Matters

Notice token = correlation_id.set(cid) followed by correlation_id.reset(token) in a finally block. The token restores the previous value when the request ends.

Under an ASGI server, worker tasks and contexts can be reused. Resetting prevents a stale ID from a finished request from bleeding into a later one that forgot to set its own. Always pair set() with reset() in middleware, and do it in finally so it runs even when the handler raises.

from contextvars import ContextVar

cv: ContextVar[str] = ContextVar("cv", default="-")

print(cv.get())          # -
token = cv.set("req-1")
print(cv.get())          # req-1
cv.reset(token)
print(cv.get())          # back to -

Surviving Background Tasks and Threads

ContextVar propagates automatically across await within the same task, but a value does not automatically follow work you push to another thread (for example run_in_executor or blocking DB drivers).

To carry the context across a thread boundary, capture it with contextvars.copy_context() and run the callable inside that copy. asyncio already does this for create_task; you must do it manually for raw executors.

import contextvars
from concurrent.futures import ThreadPoolExecutor

cid = contextvars.ContextVar("cid", default="-")

def work():
    return cid.get()

cid.set("trace-9")
ctx = contextvars.copy_context()
with ThreadPoolExecutor() as pool:
    # ctx.run carries the ContextVar value into the worker thread
    result = pool.submit(ctx.run, work).result()

print("in thread:", result)  # trace-9

Propagating Across Services

A correlation ID is only useful end-to-end if it crosses service boundaries. When your FastAPI service calls another service, forward the ID as an HTTP header so the downstream logs share the same value.

Read it from the ContextVar and inject it into every outbound client call. The receiving service's middleware reads that header instead of generating a new ID, so one ID spans the whole call chain.

import httpx
from contextvars import ContextVar

correlation_id: ContextVar[str] = ContextVar("correlation_id", default="-")

async def call_downstream(url: str):
    headers = {"X-Request-ID": correlation_id.get()}
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, headers=headers)
        return resp.json()

Putting It Together with structlog

Rather than hand-build formatters, many teams use structlog, which composes a pipeline of processors and renders JSON at the end. A processor can pull the correlation ID from the ContextVar and merge it into every event automatically.

The benefits compound: consistent JSON output, easy per-event context binding via logger.bind(...), and clean integration with the stdlib logging module so library logs are captured too.

import structlog
from contextvars import ContextVar

correlation_id: ContextVar[str] = ContextVar("correlation_id", default="-")

def add_correlation_id(logger, method_name, event_dict):
    event_dict["correlation_id"] = correlation_id.get()
    return event_dict

structlog.configure(
    processors=[
        add_correlation_id,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ]
)

correlation_id.set("abc-123")
log = structlog.get_logger()
log.info("checkout_completed", amount=49.9, currency="EUR")

Quick Check

Test your understanding of correlation ID propagation in async FastAPI services.

在交互式课程中回答

Recap

You built request-scoped, structured logging for FastAPI:

Structured JSON logs via a custom logging.Formatter (or structlog) make logs queryable.
Correlation IDs stitch every log line of one request together across functions and services.
contextvars.ContextVar holds the ID with per-request isolation and survives await boundaries.
A logging filter injects the ID onto every record so no call site must remember it.
FastAPI middleware reads X-Request-ID or generates a UUID, then pairs set() with reset(token) in finally.
Carry context into threads with copy_context() and across services by forwarding the ID header.

The result: one ID, queried in your log aggregator, reveals the complete journey of any request.

免费开始

用 AI 导师学习 FastAPI Backend Development Bootcamp — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 21
课程: 84

免费开始学习查看完整课程

常见问题解答

「结构化 JSON 日志与关联 ID」课时是免费的吗？

是的 — 「结构化 JSON 日志与关联 ID」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 FastAPI Backend Development Bootcamp 课程的其余内容，请升级到 CoddyKit PRO。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

「结构化 JSON 日志与关联 ID」这节课中我会学到什么？

输出带请求作用域关联 ID 的结构化日志，使其能够跨越异步边界和服务持续传递。你通过在浏览器中直接运行的动手代码来练习 FastAPI Backend Development Bootcamp，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 FastAPI Backend Development Bootcamp 需要有经验吗？

无需任何先前经验。CoddyKit 上的 FastAPI Backend Development Bootcamp 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 1 节课，共 4 节。

「结构化 JSON 日志与关联 ID」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 FastAPI Backend Development Bootcamp 课中编写并运行代码吗？

能。每节 FastAPI Backend Development Bootcamp 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 FastAPI Backend Development Bootcamp