Blog post example using Request ID in Async I/O Python app

example_app.py

import re
import asyncio
import logging
import logging.config
import uuid

from aiohttp import web
import aiotask_context as context


class RequestIdFilter(logging.Filter):
    def filter(self, record):
        current_task = asyncio.Task.current_task()

        if not current_task or not hasattr(current_task, 'context'):
            record.request_id = ''
            return True

        record.request_id = context.get('request_id')
        return True


logging.config.dictConfig({
    'version': 1,
    'disable_existing_loggers': True,
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'DEBUG',
            'formatter': 'default',
            'filters': ['requestid'],
        }
    },
    'filters': {
        'requestid': {
            '()': RequestIdFilter,
        },
    },
    'formatters': {
        'default': {
            'format': '%(levelname)s [%(asctime)s] %(request_id)s - %(name)s - %(message)s',
        },
    },
    'loggers': {
        None: {
            'handlers': ['console'],
            'level': 'DEBUG'
        },
        'aiohttp.access': {
            'handlers': ['console'],
            'level': 'DEBUG',
            'propagate': False
        }
    }
})

log = logging.getLogger(__name__)

REQUEST_ID_HEADER = 'X-Request-Id'
REQUEST_ID_PATTERN = re.compile('^[0-9a-z]+$')


@web.middleware
async def assign_request_id_middleware(request, handler):
    request_id = request.headers.get('x-request-id')
    if not request_id or not REQUEST_ID_PATTERN.match(request_id):
        request_id = str(uuid.uuid4()).replace('-', '')

    request['request_id'] = request_id
    context.set('request_id', request_id)
    response = await handler(request)
    response.headers[REQUEST_ID_HEADER] = request_id
    return response


async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    log.info(f"Saying hello to {name}!")
    return web.Response(text=f"Hello, {name}")


app = web.Application()
app.middlewares.append(assign_request_id_middleware)
app.add_routes([web.get('/', handle),
                web.get('/{name}', handle)])

loop = asyncio.get_event_loop()
loop.set_task_factory(context.task_factory)
web.run_app(app)