FastApi Request ID

Augments each request with unique request_id attribute and provides request id logging helpers.

Quickstart

pip install fastapi-request-id

import logging.config
from fastapi_request_id import (
    RequestContextMiddleware,
    BaseExceptionHandler,
    get_request_id,
    get_headers,
)
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from httpx import Client


class CustomExceptionHandler(BaseExceptionHandler):

    @staticmethod
    def build_response(request: Request, exc: Exception) -> Response:
        return JSONResponse(status_code=200, content={
            'error': True,
            'request_id': get_request_id(),
        })


app = FastAPI()
app.add_middleware(RequestContextMiddleware)
app.add_exception_handler(Exception, CustomExceptionHandler())

LOGGING_CONFIG = {
    'version': 1,
    'disable_existing_loggers': True,
    'filters': {
        'RequestIdFilter': {
            '()': 'fastapi_request_id.logging.RequestIdFilter',
        },
    },
    'handlers': {
        'default': {
            'level': 'INFO',
            'class': 'logging.StreamHandler',
            "filters": [
                "RequestIdFilter"
            ]
        },
    },
    'loggers': {
        '': {
            'handlers': ['default'],
            'level': 'INFO',
            'propagate': False
        },
    }
}

logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger()


@app.get('/raise')
def get_with_raise():
    raise Exception('Raised error')


@app.get('/')
def root():
    return {'request_id': get_request_id()}


@app.get('/request')
def request():
    headers = get_headers({
        'Authorization': "Bearer ...",
    })

    client = Client()
    response = client.get('http://localhost/request', headers=headers)
    return response.json()