Log context propagation in Python ASGI apps.

__init__.py

from svc import log  # noqa

log.py

import logging
import json
import time
from typing import Any


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_record = {
            "message": record.getMessage(),
            # Defaults to current time in milliseconds if not set
            "timestamp": record.__dict__.get("timestamp", int(time.time() * 1000)),
            # Defaults to empty dict if not set
            "tags": record.__dict__.get("tags", {}),
        }
        return json.dumps(log_record)


class ContextFilter(logging.Filter):
    def __init__(self) -> None:
        super().__init__()
        self.context = {}

    def set_context(self, **kwargs: Any) -> None:
        self.context.update(kwargs)

    def filter(self, record: logging.LogRecord) -> bool:
        record.tags = self.context
        return True


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

console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)  # Set the logging level for the handler

context_filter = ContextFilter()  # Set filter
console_handler.addFilter(context_filter)
logger.addFilter(context_filter)

json_formatter = JsonFormatter()  # Set formatter
console_handler.setFormatter(json_formatter)

logger.addHandler(console_handler)  # Add handler to logger

main.py

from starlette.routing import Route
import uvicorn
from starlette.applications import Starlette
from svc.middleware import LogContextMiddleware
from svc.view import view
from starlette.middleware import Middleware

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

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
import logging
from svc.log import ContextFilter


class LogContextMiddleware(BaseHTTPMiddleware):
    def __init__(self, app):
        super().__init__(app)
        self.logger = logging.getLogger()
        self.context_filter = next(
            f for f in self.logger.filters if isinstance(f, ContextFilter)
        )

    async def dispatch(self, request: Request, call_next) -> Response:
        # Extract user information from the request (headers or parameters)
        user_id = request.headers.get("Svc-User-ID", "unknown")
        platform = request.headers.get("Svc-Platform", "unknown")

        # Set context in the logger
        self.context_filter.set_context(user_id=user_id, platform=platform)

        # Log the incoming request
        self.logger.info("Handling request")

        response = await call_next(request)

        # Log the outgoing response
        self.logger.info("Finished handling request")

        # Clear context after request is handled
        self.context_filter.set_context(**{})

        return response

view.py

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


async def work() -> None:
    await asyncio.sleep(1)
    logging.info("Work done after 1 second")


async def view(request: Request) -> JSONResponse:
    await work()

    return JSONResponse({"message": "Did some work!"})