Mastering FastAPI Async Logging: Structured, Rotating, and Worker‑Ready Logs for High‑Performance Apps

Introduction

When you start building production‑grade APIs with FastAPI async logging, the first thing you’ll notice is that the default print statements quickly become a liability. As your service scales behind Uvicorn or Gunicorn workers, you need a logging strategy that respects the asynchronous nature of FastAPI, works across multiple processes, and gives you structured data you can ship to ELK, Loki, or any observability platform.

In this post we’ll walk through a complete, hands‑on setup:

• Configuring the built‑in logging module for FastAPI.
• Picking async‑compatible loggers such as structlog and loguru.
• Emitting logs from background tasks and WebSocket connections.
• Making FastAPI logs play nicely with Uvicorn/Gunicorn workers.
• Implementing log rotation and structured logging for async apps.

By the end you’ll have a ready‑to‑run codebase that you can drop into any FastAPI project and start collecting meaningful logs immediately.

Configuring Standard Logging in FastAPI

FastAPI itself does not ship its own logger; it relies on Python’s standard logging library. The good news is that logging works perfectly fine in async code as long as you don’t block the event loop with heavy I/O inside a formatter.

1. Create a logging configuration

Place a logging_config.py file in your project:

import logging
import sys
from logging.config import dictConfig

LOG_LEVEL = "INFO"

LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "standard": {
            "format": "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S",
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "stream": sys.stdout,
            "formatter": "standard",
        },
    },
    "loggers": {
        "uvicorn.error": {"level": LOG_LEVEL, "handlers": ["console"], "propagate": False},
        "uvicorn.access": {"level": LOG_LEVEL, "handlers": ["console"], "propagate": False},
        "fastapi": {"level": LOG_LEVEL, "handlers": ["console"], "propagate": False},
        # Your application logger
        "app": {"level": LOG_LEVEL, "handlers": ["console"], "propagate": False},
    },
    "root": {"level": LOG_LEVEL, "handlers": ["console"]},
}

def configure_logging():
    dictConfig(LOGGING_CONFIG)

2. Wire it into the FastAPI lifespan

from fastapi import FastAPI
from logging_config import configure_logging
import logging

logger = logging.getLogger("app")

def create_app() -> FastAPI:
    app = FastAPI()

    @app.on_event("startup")
    async def startup_event():
        configure_logging()
        logger.info("🚀 FastAPI app started with async logging enabled")

    @app.get("/ping")
    async def ping():
        logger.debug("Ping endpoint hit")
        return {"msg": "pong"}

    return app

app = create_app()

The configure_logging call runs once per process (worker). Because we used disable_existing_loggers=False, any libraries that create their own loggers (SQLAlchemy, HTTPX, etc.) will inherit the same format and level.

Why this works for async: The StreamHandler writes to sys.stdout which is non‑blocking for most production containers (Docker, Kubernetes). If you later switch to a file handler, you’ll need an async‑aware handler – we’ll cover that later.

Choosing Async‑Compatible Loggers (structlog, loguru) for FastAPI

While the standard library is fine, modern async applications benefit from structured logs and a richer API. Two popular choices are structlog and loguru. Both can be used without sacrificing async performance.

1. structlog – the “structured” champion

structlog decorates the standard logger, turning dicts into JSON (or any format) without blocking the event loop.

import logging
import sys
import structlog
from logging.config import dictConfig

def configure_structlog():
    LOG_LEVEL = "INFO"
    dictConfig({
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "json": {
                "format": "%(message)s",
                "class": "pythonjsonlogger.jsonlogger.JsonFormatter",
            },
        },
        "handlers": {
            "default": {
                "class": "logging.StreamHandler",
                "formatter": "json",
                "stream": sys.stdout,
            },
        },
        "loggers": {
            "": {"handlers": ["default"], "level": LOG_LEVEL},
        },
    })

    structlog.configure(
        processors=[
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer(),
        ],
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

Usage inside FastAPI:

# main.py (continued)
from structlog_config import configure_structlog
import structlog

log = structlog.get_logger("app")

@app.on_event("startup")
async def startup_event():
    configure_structlog()
    log.info("application_start", env="production")

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    log.debug("fetch_user", user_id=user_id)
    # pretend we query a DB...
    return {"user_id": user_id}

The logs will look like:

{
  "timestamp":"2026-07-01T12:34:56.789Z",
  "event":"fetch_user",
  "user_id":42,
  "level":"debug"
}

Because structlog ultimately uses the standard logging handlers, it works in any async context, and the JSON output is ready for Loki or Elasticsearch.

2. loguru – “the friendly logger”

loguru provides a drop‑in replacement for the logging module with sensible defaults and built‑in rotation. It is not async‑aware out of the box, but its I/O is performed in a background thread, which means it does not block the event loop.

from loguru import logger
import sys

def configure_loguru():
    logger.remove()  # remove the default sink
    logger.add(
        sys.stdout,
        format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level:<8} | {message}",
        level="INFO",
        backtrace=True,
        diagnose=True,
        enqueue=True,          # <-- crucial: sends logs to a thread‑safe queue
    )

Usage:

# main.py (continued)
from loguru_config import configure_loguru

@app.on_event("startup")
async def startup_event():
    configure_loguru()
    logger.info("FastAPI with loguru ready")

enqueue=True ensures that log records are processed in a separate thread, making it safe for async code. If you later need file rotation, loguru can handle it with a single line.

Logging from Background Tasks and WebSockets

FastAPI’s async nature means you’ll often launch background tasks (BackgroundTasks) or long‑living WebSocket connections. Both need careful logging to avoid losing context.

1. Background tasks

from fastapi import BackgroundTasks, FastAPI
import structlog

log = structlog.get_logger("app")

app = FastAPI()

def heavy_computation(item_id: int):
    log.info("background_task_start", item_id=item_id)
    # Simulate CPU‑bound work
    import time
    time.sleep(2)  # blocking is okay here because it's a separate thread
    log.info("background_task_done", item_id=item_id)

@app.post("/process/{item_id}")
async def process(item_id: int, background_tasks: BackgroundTasks):
    background_tasks.add_task(heavy_computation, item_id)
    log.info("request_received", endpoint="/process", item_id=item_id)
    return {"status": "queued"}

Even though heavy_computation blocks, it runs in the thread pool managed by FastAPI’s background task system, so the main event loop stays responsive. The logger retains its context because we bind the item_id directly in each call.

2. WebSocket connections

WebSockets keep a single coroutine alive for the whole connection. Use a per‑connection logger bound with the client’s identifier.

from fastapi import WebSocket, WebSocketDisconnect, FastAPI
import structlog

app = FastAPI()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await websocket.accept()
    log = structlog.get_logger("app").bind(client_id=client_id)
    log.info("ws_connected")
    try:
        while True:
            data = await websocket.receive_text()
            log.debug("ws_message_received", payload=data)
            # Echo back
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        log.info("ws_disconnected")

Because structlog creates a new bound logger for each connection, you get clean, per‑client logs without manually passing the identifier around.

Integrating FastAPI Logs with Uvicorn/Gunicorn Workers

When you serve FastAPI behind Uvicorn or Gunicorn (with uvicorn.workers.UvicornWorker), each worker is a separate process. You need to ensure that:

1. Loggers are configured per process. The startup event runs once per worker, so our earlier configure_logging or configure_structlog calls are sufficient.
2. Access logs are captured. Uvicorn emits its own access logs; you can forward them to your structured logger.

1. Forwarding Uvicorn access logs to structlog

Create a small wrapper:

import structlog
import sys
import logging
from logging import Handler

class StructlogHandler(Handler):
    def emit(self, record):
        # Convert LogRecord to dict and pass to structlog
        log = structlog.get_logger(record.name)
        log_msg = self.format(record)
        log.msg(event=log_msg, level=record.levelname.lower())

def configure_uvicorn_structlog():
    structlog.configure(
        processors=[
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer(),
        ],
        logger_factory=structlog.stdlib.LoggerFactory(),
    )
    handler = StructlogHandler()
    handler.setFormatter(logging.Formatter("%(message)s"))
    logging.getLogger("uvicorn.access").addHandler(handler)
    logging.getLogger("uvicorn.error").addHandler(handler)

Add this in your main.py startup:

from uvicorn_structlog import configure_uvicorn_structlog

@app.on_event("startup")
async def startup_event():
    configure_structlog()
    configure_uvicorn_structlog()
    logger.info("All loggers (app + uvicorn) are now structured")

Now every request line like GET /ping 200 OK appears as a JSON object with timestamps and request IDs.

2. Running with Gunicorn

gunicorn -k uvicorn.workers.UvicornWorker \
    -w 4 \
    -b 0.0.0.0:8000 \
    --log-level info \
    --access-logfile - \
    main:app

The --access-logfile - tells Gunicorn to pipe access logs to stdout, where our StructlogHandler will capture them. Because each worker runs the FastAPI startup routine, logs are correctly bound to the right process ID (process field can be added via a processor if you need it).

Best Practices for Log Rotation and Structured Logging in Async Apps

1. Rotate with loguru (simple)

from loguru import logger

logger.add(
    "logs/app_{time:%Y-%m-%d}.log",
    rotation="00:00",          # daily rotation at midnight
    retention="14 days",      # keep two weeks
    compression="zip",       # compress old logs
    enqueue=True,             # thread‑safe for async
)

2. Rotate with logging.handlers.TimedRotatingFileHandler (standard)

import logging
from logging.handlers import TimedRotatingFileHandler

handler = TimedRotatingFileHandler(
    filename="logs/app.log",
    when="midnight",
    backupCount=14,
    encoding="utf-8",
)
handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
logging.getLogger("app").addHandler(handler)

Both approaches are safe for async because the handlers write in a separate thread (the standard library does this internally).

3. Include request IDs for traceability

FastAPI’s Request object has a unique state you can populate with a UUID in a middleware:

import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
import structlog

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        request.state.request_id = request_id
        # Bind request_id to structlog for the duration of the request
        structlog.contextvars.bind_contextvars(request_id=request_id)
        response = await call_next(request)
        structlog.contextvars.unbind_contextvars("request_id")
        response.headers["X-Request-ID"] = request_id
        return response

app.add_middleware(RequestIDMiddleware)

Now every log entry automatically contains request_id, making it trivial to correlate logs across background tasks, websockets, and external services.

4. Avoid blocking I/O in formatters

Never perform network calls or heavy file parsing inside a formatter. Keep the formatter pure (e.g., JSON rendering). If you need to enrich logs with data from a DB, do it before calling logger.info(...).

5. Test your logging in the async context

Use pytest-asyncio to spin up a test client and assert that logs contain expected fields:

import pytest
from httpx import AsyncClient
from main import app

@pytest.mark.asyncio
async def test_logging(caplog):
    async with AsyncClient(app=app, base_url="http://test") as client:
        await client.get("/ping")
    # caplog captures standard logging output
    assert any("ping" in rec.message.lower() for rec in caplog.records)

Putting It All Together – A Minimal Template

from fastapi import FastAPI
from structlog_config import configure_structlog
from uvicorn_structlog import configure_uvicorn_structlog
import structlog

log = structlog.get_logger("app")

def create_app() -> FastAPI:
    app = FastAPI()

    @app.on_event("startup")
    async def startup():
        configure_structlog()
        configure_uvicorn_structlog()
        log.info("fastapi_async_logging_ready")

    # Middleware for request IDs
    from starlette.middleware.base import BaseHTTPMiddleware
    from starlette.requests import Request
    import uuid, structlog.contextvars as ctx

    class RequestIDMiddleware(BaseHTTPMiddleware):
        async def dispatch(self, request: Request, call_next):
            request_id = str(uuid.uuid4())
            ctx.bind_contextvars(request_id=request_id)
            response = await call_next(request)
            ctx.unbind_contextvars("request_id")
            response.headers["X-Request-ID"] = request_id
            return response

    app.add_middleware(RequestIDMiddleware)

    @app.get("/ping")
    async def ping():
        log.debug("ping_endpoint")
        return {"msg": "pong"}

    # Example WebSocket
    @app.websocket("/ws/{client_id}")
    async def ws_endpoint(websocket: WebSocket, client_id: str):
        await websocket.accept()
        ws_log = log.bind(client_id=client_id)
        ws_log.info("ws_connected")
        try:
            while True:
                data = await websocket.receive_text()
                ws_log.debug("message_received", payload=data)
                await websocket.send_text(f"ECHO: {data}")
        except WebSocketDisconnect:
            ws_log.info("ws_disconnected")

    return app

app = create_app()

Run it with:

uvicorn app:app --host 0.0.0.0 --port 8000 --workers 2

Or, for production:

gunicorn -k uvicorn.workers.UvicornWorker -w 4 app:app

All logs – request logs, background tasks, and websockets – are now fastapi async logging‑compatible, structured, and ready for rotation.

Further Reading

If you’re interested in deeper async patterns, check out our guide on FastAPI Async vs Sync – When to Use Each, Benchmarks, and Real‑World Tips.
For lifecycle nuances that affect logging (e.g., when to configure in lifespan vs. startup), see FastAPI Lifespan vs Startup Events: The Mistake That Breaks Async Apps.

Key Takeaways

• Standard logging works for FastAPI async apps, but structured loggers (structlog, loguru) give you JSON output and easier correlation.
• Configure loggers in a startup/lifespan event so each Uvicorn/Gunicorn worker gets its own correctly initialized logger.
• Background tasks and WebSockets should use bound loggers (or contextvars) to keep request‑level metadata like request_id.
• Forward Uvicorn access logs to your chosen logger to have a single source of truth for request tracing.
• Log rotation can be handled by loguru’s built‑in rotation or by TimedRotatingFileHandler; both are safe for async code when enqueue=True or the handler runs in a background thread.
• Adding request IDs via middleware dramatically improves observability across async boundaries.

With these patterns in place, your FastAPI services will generate clean, non‑blocking, and searchable logs that scale with the number of workers and the complexity of your async workflows. Happy logging!