starlette-context

The middleware is responsible for creating and managing the request context. You must choose one of the two available implementations based on your application's needs:

• ContextMiddleware: Built on Starlette's BaseHTTPMiddleware. It is simpler to use and suitable for most standard web applications.
• RawContextMiddleware: Operates at a lower ASGI level. Use this if your application relies heavily on StreamingResponse or handles very large responses, as it avoids memory issues and provides better performance for streaming.

Both middlewares support the same plugins configuration.

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette_context.middleware import ContextMiddleware
from starlette_context import plugins

middleware = [
    Middleware(
        ContextMiddleware,
        plugins=(
            plugins.RequestIdPlugin(),
            plugins.CorrelationIdPlugin()
        )
    )
]

app = Starlette(middleware=middleware)

The Problem

When using ContextMiddleware, background tasks execute after the middleware's context manager has already exited and reset the context. While Python's ContextVar inheritance (PEP 567) might make context appear available in background tasks, this is an implementation detail and is not guaranteed.

The Recommended Pattern: Explicit Context Passing

To ensure reliability, you must capture a copy of the context data during the request and pass it explicitly as an argument to your background task function.

Do not attempt to call context.get() directly inside a background task function.

from fastapi import FastAPI, BackgroundTasks
from starlette_context import context
from starlette_context.middleware import ContextMiddleware

app = FastAPI()
app.add_middleware(ContextMiddleware)

def process_item(item_id: str, context_data: dict):
    # ✅ RECOMMENDED: Use explicitly passed context data
    # This is reliable and guaranteed to work
    print(f"Processing item {item_id} with context: {context_data}")
    request_id = context_data.get("X-Request-ID")

@app.post("/items/{item_id}")
async def create_item(item_id: str, background_tasks: BackgroundTasks):
    # Capture context data during request
    context_data = context.data.copy()

    # Pass context data explicitly to the background task
    background_tasks.add_task(process_item, item_id, context_data)

    return {"message": "Item will be processed"}

To use starlette-context, add ContextMiddleware to your Starlette middleware stack. You can pass plugins to the middleware to automatically populate the context with data like CorrelationIdPlugin or RequestIdPlugin. Once the middleware is active, you can access and modify the request-scoped context using the context object within your routes or other middleware.

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette_context import context, plugins
from starlette_context.middleware import ContextMiddleware

# ... setup routes ...

async def index(request):
    # Store data in the context
    context["custom_value"] = "This will be visible in logs"
    return JSONResponse(context.data)

app = Starlette(
    routes=[Route("/", index)],
    middleware=[
        Middleware(
            ContextMiddleware,
            plugins=(
                plugins.CorrelationIdPlugin(),
                plugins.RequestIdPlugin(),
            ),
        ),
    ],
)

You can use starlette-context to automatically inject request-specific data into your logs. When using a logging library like structlog, create a processor that checks if a context exists before attempting to merge its data.

Important: Always use the context.exists() guard. If you attempt to access context.data outside of a request cycle (such as during application startup), it will raise a ContextDoesNotExistError.

import structlog
from starlette_context import context

def add_context(logger, method_name, event_dict):
    """Structlog processor that merges context data into every log entry."""
    if context.exists():
        event_dict.update(context.data)
    return event_dict

structlog.configure(
    processors=[
        add_context,
        structlog.dev.ConsoleRenderer(),
    ]
)

logger = structlog.get_logger()

# Inside a route
@app.route("/")
async def index(request):
    logger.info("Processing request")
    return JSONResponse({"message": "Hello World"})

When writing custom FastAPI exception handlers, you should safely check if the context exists before attempting to access it. Use context.exists() or wrap your access in a try/except block catching ContextDoesNotExistError to prevent errors when an exception occurs outside of a valid request cycle.

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from starlette_context import context
from starlette_context.middleware import ContextMiddleware
from starlette_context.errors import ContextDoesNotExistError

app = FastAPI()
app.add_middleware(ContextMiddleware)

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    # Safely access context in exception handler
    extra_data = {}
    try:
        if context.exists():
            extra_data = {"request_id": context.get("X-Request-ID")}
    except ContextDoesNotExistError:
        pass

    return JSONResponse(
        status_code=exc.status_code,
        content={
            "message": exc.detail,
            **extra_data
        }
    )