Log context propagation in Python ASGI apps. Details here: https://rednafi.com/python/log_context_propagation/

__init__.py

from svc import log  # noqa

log.py

# log.py

import contextvars
import json
import logging
import time

# Set up the context variable with default values
default_context = {"user_id": "unknown", "platform": "unknown"}
log_context_var = contextvars.ContextVar("log_context", default=default_context.copy())


# Custom log formatter
class ContextAwareJsonFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "message": record.getMessage(),
            # Add millisecond precision timestamp
            "timestamp": int(time.time() * 1000),
            # Get the context from the context variable in a concurrent-safe way
            "tags": log_context_var.get(),
        }
        return json.dumps(log_data)


# Set up the logger
logger = logging.getLogger()
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
formatter = ContextAwareJsonFormatter()
handler.setFormatter(formatter)
logger.addHandler(handler)

main.py

# main.py

import uvicorn
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.routing import Route

from svc.middleware import LogContextMiddleware
from svc.view import view

middlewares = [Middleware(LogContextMiddleware)]

app = Starlette(routes=[Route("/", view)], middleware=middlewares)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

middleware.py

# middleware.py

import logging
from collections.abc import Awaitable, Callable

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

from svc.log import default_context, log_context_var


# Middleware for setting log context
class LogContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(
        self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
    ) -> Response:
        context = default_context.copy()
        user_id = request.headers.get("Svc-User-Id")
        platform = request.headers.get("Svc-Platform")

        if user_id:
            context["user_id"] = user_id
        if platform:
            context["platform"] = platform

        token = log_context_var.set(context)

        try:
            logging.info("From middleware: request started")
            response = await call_next(request)
            logging.info("From middleware: request ended")
        finally:
            # Reset the context after the request is processed
            log_context_var.reset(token)

        return response

view.py

# view.py

import asyncio
import logging

from starlette.requests import Request
from starlette.responses import JSONResponse


async def view(request: Request) -> JSONResponse:
    await work()
    logging.info("From view function: work finished")
    return JSONResponse({"message": f"Work work work!!!"})


async def work() -> None:
    logging.info("From work function: work started")
    await asyncio.sleep(1)

This doesn't consider concurrency at all. Try increasing the asyncio.sleep time, and making two requests in parallel:

http localhost:8000 "Svc-User-ID: u1"
http localhost:8000 "Svc-User-ID: u2"

You'll get:

INFO:     Started server process [303329]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
{"message": "Handling request", "timestamp": 1723136966847, "tags": {"user_id": "u1", "platform": "unknown"}}
{"message": "Handling request", "timestamp": 1723136969232, "tags": {"user_id": "u2", "platform": "unknown"}}
{"message": "Work done after 1 second", "timestamp": 1723136976848, "tags": {"user_id": "u2", "platform": "unknown"}}
{"message": "Finished handling request", "timestamp": 1723136976849, "tags": {"user_id": "u2", "platform": "unknown"}}
INFO:     127.0.0.1:54790 - "GET / HTTP/1.1" 200 OK
{"message": "Work done after 1 second", "timestamp": 1723136979232, "tags": {"user_id": "u2", "platform": "unknown"}}
{"message": "Finished handling request", "timestamp": 1723136979233, "tags": {"user_id": "u2", "platform": "unknown"}}
INFO:     127.0.0.1:54796 - "GET / HTTP/1.1" 200 OK

@selimb Aaah. You’re correct. The logger filter is shared across requests and that bleeds the context. Need to play around with contextvars to see how to eliminate that. Will update the code and the post soon. Thanks for reporting.

Used contextvars to prevent context leak across requests. Now making multiple requests looks like this:

request:

for i in {1..2}; do
  curl -H "Svc-User-Id: 100${i}" -H "Svc-Platform: web" "http://localhost:8000" &
done
wait

response:

INFO:     Started server process [44703]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
`{"message": "From middleware: request started", "timestamp": 1723168855822, "tags": {"user_id": "1002", "platform": "web"}}
{"message": "From middleware: request started", "timestamp": 1723168855822, "tags": {"user_id": "1001", "platform": "web"}}
{"message": "From work function: work started", "timestamp": 1723168855822, "tags": {"user_id": "1002", "platform": "web"}}
{"message": "From work function: work started", "timestamp": 1723168855822, "tags": {"user_id": "1001", "platform": "web"}}
{"message": "From view function: work finished", "timestamp": 1723168856823, "tags": {"user_id": "1002", "platform": "web"}}
{"message": "From view function: work finished", "timestamp": 1723168856823, "tags": {"user_id": "1001", "platform": "web"}}
{"message": "From middleware: request ended", "timestamp": 1723168856823, "tags": {"user_id": "1002", "platform": "web"}}
{"message": "From middleware: request ended", "timestamp": 1723168856823, "tags": {"user_id": "1001", "platform": "web"}}
INFO:     127.0.0.1:54902 - "GET / HTTP/1.1" 200 OK

👍 Good stuff!