FastAPI is a Python framework for building APIs. Uvicorn is the server that runs those applications through ASGI, Python’s interface between asynchronous web servers and applications. A FastAPI logger usually uses Python’s logging machinery; application events, Uvicorn server messages and HTTP access records still come from different loggers.
Giving them the same JSON format solves only part of the problem. To connect an application error to a request, each event also needs a request ID. That ID must stay with the right request while other requests run. FastAPI adds an important boundary: ordinary def endpoints and dependencies run in a thread pool, while async def endpoints run asynchronously. FastAPI documents this distinction.
The example below uses one JSON formatter and one output handler. Pure ASGI middleware generates a request ID, returns it in a response header and clears request context afterward. It also replaces Uvicorn’s default access lines with a small request summary that excludes URLs, query strings and headers. The concurrency check shows which context survives a thread-pool call and which changes stay inside it.
For logging levels, exception handling and general Python configuration, see Python logging best practices. For a smaller introduction to the standard library, start with the Python logging example.
Configure and verify FastAPI request logging
These steps apply to FastAPI applications running under Uvicorn, including applications with both synchronous and asynchronous handlers. They check formatting, request correlation and duplicate emission locally. They do not establish distributed tracing or prove that a collector stored every event.
The example was tested with Python 3.13.5, FastAPI 0.162.0, Starlette 1.10.0, Uvicorn 0.54.0, structlog 26.1.0 and HTTPX 0.28.1. Keep the test when changing those versions or adding middleware.
Give JSON formatting one owner
Install the tested dependencies in a virtual environment:
python3 -m venv .venv
.venv/bin/python -m pip install fastapi==0.162.0 starlette==1.10.0 uvicorn==0.54.0 structlog==26.1.0 httpx==0.28.1
Save this as log_setup.py:
import logging.config
import structlog
shared = [
structlog.contextvars.merge_contextvars,
structlog.stdlib.add_log_level,
structlog.stdlib.add_logger_name,
structlog.processors.TimeStamper(fmt="iso", utc=True),
]
def configure_logging():
formatter = structlog.stdlib.ProcessorFormatter(
foreign_pre_chain=shared,
processors=[
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer(),
],
)
logging.config.dictConfig({
"version": 1,
"disable_existing_loggers": False,
"formatters": {"json": {"()": lambda: formatter}},
"handlers": {"stdout": {
"class": "logging.StreamHandler",
"stream": "ext://sys.stdout",
"formatter": "json",
}},
"root": {"handlers": ["stdout"], "level": "INFO"},
"loggers": {
name: {
"handlers": [], "level": "INFO", "propagate": True,
}
for name in (
"checkout", "uvicorn", "uvicorn.error", "uvicorn.access"
)
},
})
structlog.configure(
processors=shared + [
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
ProcessorFormatter accepts both structlog events and ordinary Python logging records. Structlog passes an event dictionary to it; the formatter renders JSON once. Ordinary records use foreign_pre_chain to acquire the same timestamp, level and logger fields. This follows the mechanisms in structlog’s standard-library integration documentation.
Only the root logger owns an output handler. checkout and uvicorn.error pass records upward to that handler. Attaching another output handler to either child while leaving propagation enabled can print the same record twice. Python’s logging reference explains propagation and duplicate handling.
The launcher below disables uvicorn.access after this configuration runs. That order matters: applying another logging configuration afterward can undo the access-log setting.
Bind request context before entering FastAPI
Save this as app.py:
import asyncio
import time
from uuid import uuid4
import structlog
from fastapi import Depends, FastAPI
from fastapi.responses import JSONResponse
log = structlog.get_logger("checkout")
api = FastAPI()
def bind_in_worker():
structlog.contextvars.bind_contextvars(
worker_note="dependency-only",
)
log.info("dependency_checked")
@api.get("/async", dependencies=[Depends(bind_in_worker)])
async def async_check():
await asyncio.sleep(0.05)
log.info("async_checked")
return {"ok": True}
@api.get("/sync")
def sync_check():
time.sleep(0.05)
log.info("sync_checked")
return {"ok": True}
@api.get("/handled")
async def handled_check():
log.error(
"inventory_unavailable", error_code="demo_unavailable",
)
return JSONResponse({"detail": "Demo failure"}, status_code=503)
class RequestLogs:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
return await self.app(scope, receive, send)
structlog.contextvars.clear_contextvars()
request_id = uuid4().hex
structlog.contextvars.bind_contextvars(
request_id=request_id, service="checkout-api",
)
status = 500
started = time.perf_counter()
async def send_with_id(message):
nonlocal status
if message["type"] == "http.response.start":
status = message["status"]
message = dict(message)
message["headers"] = list(message.get("headers", [])) + [
(b"x-request-id", request_id.encode("ascii")),
]
await send(message)
try:
await self.app(scope, receive, send_with_id)
finally:
try:
log.info(
"request_finished",
method=scope["method"],
status=status,
duration_ms=round(
(time.perf_counter() - started) * 1000, 2,
),
)
finally:
structlog.contextvars.clear_contextvars()
app = RequestLogs(api)
The request ID and response status are local to each middleware call. They are not stored on the shared middleware instance. The outer wrapper also keeps context active while FastAPI sends a handled error response.
This uses pure ASGI middleware rather than BaseHTTPMiddleware. Starlette documents a limitation in BaseHTTPMiddleware that prevents endpoint changes to context variables from propagating upward. Pure ASGI middleware avoids that particular boundary; it does not remove thread-pool boundaries. See Starlette’s middleware documentation.
The example generates IDs locally. If you later accept an incoming request ID from a trusted gateway, validate its length and characters. Do not copy arbitrary header values into logs. A request ID is a correlation value, not proof of identity.
duration_ms measures how long the wrapped application call takes. It can include response streaming and background work. It is not the client’s measured latency. Likewise, the recorded status is the response-start status: a later streaming failure can follow a 200, and this summary does not prove successful delivery to the client.
Start Uvicorn without replacing the configuration
Save this as run.py:
import uvicorn
from log_setup import configure_logging
if __name__ == "__main__":
configure_logging()
uvicorn.run(
"app:app",
host="127.0.0.1",
port=8000,
log_config=None,
access_log=False,
)
Start it from the directory containing these files:
.venv/bin/python run.py > events.jsonl 2> server.stderr
Use this launcher for the example. It configures logging before Uvicorn starts, preserves that configuration with log_config=None and disables default access logging with access_log=False. The request_finished event supplies the request summary instead.
Uvicorn also supports external JSON or YAML logging configuration and a separate access-log switch. These are documented in Uvicorn’s settings reference. If you move configuration into a process manager or add worker processes, check the initialization order again.
Disabling access lines here avoids collecting their URL and query-string content. If you retain them elsewhere, review what they contain. An access line and an application error for the same request are separate events; their coexistence alone is not duplicate emission.
Check overlapping requests and exact event counts
Save this as check.py. Run it once against a freshly started server so events.jsonl contains only startup messages and this test’s requests.
import asyncio
import json
from collections import Counter
from pathlib import Path
import httpx
async def main():
async with httpx.AsyncClient(
base_url="http://127.0.0.1:8000",
) as client:
replies = await asyncio.gather(*(
client.get(path)
for path in ("/async", "/sync", "/handled")
))
assert [r.status_code for r in replies] == [200, 200, 503]
ids = [r.headers["x-request-id"] for r in replies]
assert len(set(ids)) == 3
await asyncio.sleep(0.1)
rows = [
json.loads(line)
for line in Path("events.jsonl").read_text().splitlines()
]
assert not any(r.get("logger") == "uvicorn.access" for r in rows)
app_rows = [r for r in rows if r.get("logger") == "checkout"]
assert len(app_rows) == 7
assert {r["request_id"] for r in app_rows} == set(ids)
expected = [
Counter(
dependency_checked=1, async_checked=1, request_finished=1,
),
Counter(sync_checked=1, request_finished=1),
Counter(inventory_unavailable=1, request_finished=1),
]
for request_id, counts in zip(ids, expected):
group = [r for r in app_rows if r["request_id"] == request_id]
assert Counter(r["event"] for r in group) == counts
assert all(r["service"] == "checkout-api" for r in group)
for row in group:
assert ("worker_note" in row) == (
row["event"] == "dependency_checked"
)
print("PASS: 7 application records, 3 request IDs, expected context boundaries")
asyncio.run(main())
In another terminal:
.venv/bin/python check.py
The tested run returned:
PASS: 7 application records, 3 request IDs, expected context boundaries
The seven application records are three for /async, two for /sync and two for /handled. Startup messages are additional records from uvicorn.error; the test deliberately excludes them from the application count. Parsing every line also catches accidental plain-text output.
Understand the context boundary before adding more fields
In the tested run, the request ID bound by asynchronous middleware reached both the synchronous dependency and the synchronous endpoint. However, worker_note, bound inside the synchronous dependency, appeared only on dependency_checked. It did not return to the asynchronous endpoint or its request summary.
These synthetic excerpts show selected fields from that behavior. They omit timestamps and other fields for clarity:
{"event":"dependency_checked","request_id":"23ef736c6bc74f049e560b163d735d2a","worker_note":"dependency-only","service":"checkout-api"}
{"event":"async_checked","request_id":"23ef736c6bc74f049e560b163d735d2a","service":"checkout-api"}
{"event":"sync_checked","request_id":"f435c46cf71849c6aca3d32df04b4a86","service":"checkout-api"}
IDs, timestamps and event order vary between runs. The assertions check relationships rather than a fixed ordering.
Context variables hold values for the current execution context. Structlog warns that synchronous and asynchronous contexts can be isolated in hybrid applications such as FastAPI. That does not mean every value disappears at every boundary. Its context-variable documentation explains the mechanism and request-entry cleanup.
Bind shared request fields before crossing into synchronous code. If a dependency discovers a value that the caller needs, return it explicitly and bind it in the caller’s context. Do not rely on a worker’s context changes returning automatically.
A child task can also retain a copied context after the request ends. Clearing the middleware’s context does not clear every task it previously created. Pass deliberate correlation fields to background jobs and test those jobs separately.
Separate handled failures from server exceptions
The /handled route deliberately emits one error event and returns 503. It does not raise an unhandled exception. That distinction keeps the seven-record assertion precise.
Unhandled exceptions follow an additional server logging path. In this wrapper, cleanup runs before an exception reaches Uvicorn, so do not assume the later uvicorn.error traceback has the request ID. Startup messages also have no request ID because no request exists yet.
If you add request-scoped exception logging, test one controlled failure in a local environment. Check the application exception record, request summary and Uvicorn record separately. Decide which traceback to retain rather than suppressing all server errors to make the count smaller.
Tracebacks and exception messages can contain secrets or personal data. Protect log access and retention, and avoid capturing request bodies, cookies, authorization headers or complete header dictionaries. The short sleep calls and demonstration failure routes above belong in the local example, not a production API.
Verify the same events after Docker collection
For containerized applications, keep JSON on stdout and use the documented Docker-to-Fluxtail collection path. It uses Fluent Bit to read Docker JSON log files, add container details and send batches to an HTTP JSON receiver.
Create the receiver and copy its exact URL. Use a receiver-bound token with ingest:write, sent as a Bearer token. Keep that token out of source control and application log messages. Fluent Bit takes the receiver host, port and path separately; preserve all parts of the copied URL.
The documented approach mounts Docker’s log directory read-only, avoids the Docker socket and uses a filesystem retry buffer. Check that your containers actually use the expected Docker JSON-file path before applying it.
Docker wraps application output in an outer record. Verify that the inner application JSON is parsed before relying on its fields. In particular, this example emits severity as level and message text as event; do not assume those names automatically become Fluxtail’s severity and message fields. The Docker integration defaults stdout to INFO when severity is absent, so an unmapped level: error could otherwise look informational.
To check the collected result:
- Send a fresh test request and record its returned
x-request-id. - Open the receiver’s stream in Live Tail and select the test’s time window.
- Find the corresponding application event and inspect the stored row. Compare its request ID and
checkout-apiservice value with the local JSON. - Check the handled failure’s severity. Confirm that
errorwas preserved or deliberately mapped, rather than replaced by the stdout fallback. - Compare the expected application events per request. Check for nested JSON, dropped fields or duplicate collection before creating service or severity filters.
Fluxtail’s Live Tail documentation describes inspecting retained rows and verifying a sender with a unique marker. A visible row establishes that the particular event was stored and is readable. It does not establish complete delivery for every request, automatic request-ID filtering or distributed tracing.