aiohttp contextvars logging

decorators.py

from __future__ import annotations

import logging
from functools import wraps
from typing import Callable, TYPE_CHECKING

if TYPE_CHECKING:
    from aiohttp import web

logger = logging.getLogger(__name__)


def authorization_required() -> Callable:
    """Extract user authorization from request."""

    def decorator(func: Callable) -> Callable:
        """Decorate handler function."""

        @wraps(func)
        async def wrapper(request: web.Request, *args, **kwargs) -> web.Response:
            """Inject user authorization data into context."""
            json_data = await request.json()

            ctx = request.app["ctx"]
            ctx["user_id"].set(json_data["user_id"])

            return await func(request, *args, **kwargs)

        return wrapper

    return decorator

handlers.py

import logging

from aiohttp import web

from decorators import authorization_required

logger = logging.getLogger("web-bot")


@authorization_required()
async def simple_handler(request: web.Request) -> web.Response:
    logger.info("Someone called simple handler")
    return web.Response(text="OK")

json_logging.py

import logging
from contextvars import ContextVar

from aiohttp.web import Application
from pythonjsonlogger import jsonlogger


class CustomJsonFormatter(jsonlogger.JsonFormatter):

    def __init__(self, ctx: dict[str, ContextVar], *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._ctx = ctx

    def add_fields(self, log_record, record, message_dict):
        super().add_fields(log_record, record, message_dict)

        for key in self._ctx:
            if value := self._ctx[key].get():
                log_record[key] = value


def setup_logging(app: Application) -> None:
    """Setup logging for aiohttp app."""
    try:
        context = app["ctx"]
    except KeyError:
        msg = "ContextVars should be registered before calling setup_logging"
        raise RuntimeError(msg)

    root_logger = logging.getLogger()
    root_logger.handlers.clear()

    format_str = '%(asctime) %(message) %(levelname) %(name)'
    formatter = CustomJsonFormatter(context, format_str)

    log_handler = logging.StreamHandler()
    log_handler.setFormatter(formatter)
    root_logger.addHandler(log_handler)

main.py

import logging
from contextvars import ContextVar

from aiohttp import web

from handlers import simple_handler
from json_logging import setup_logging


def create_app() -> web.Application:
    """Create and configure an instance of the aiohttp application."""
    app = web.Application()

    register_ctx_vars(app)
    setup_logging(app)
    register_routes(app)

    return app


def register_routes(app: web.Application) -> None:
    """Register routes for aiohttp application."""
    routes = [
        web.route('POST', '/', simple_handler),
    ]
    app.add_routes(routes)


def register_ctx_vars(app: web.Application) -> None:
    """Register context variables for aiohttp application."""
    app["ctx"] = {
        "user_id": ContextVar("user_id", default=None),
        "chat_id": ContextVar("chat_id", default=None),
    }


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)

    app = create_app()
    web.run_app(app, access_log=None)

requirements.txt

aiohttp==3.8.6
python-json-logger==2.0.7

PEP: https://peps.python.org/pep-0550/

aiohttp contextvars: https://docs.aiohttp.org/en/v3.8.1/web_advanced.html#contextvars-support