How to log operation id and status for each route?

[ojii]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

@app.middleware("http")
async def log(request, call_next):
    response = call_next(request)
    if response.status_code >= 300:
        return response
    operation_id = get_operation_id_somehow() # is this possible?
    print("LOG", operation_id, get_access_token(request))
    return response

Description

I want to log the Operation ID, as well as the access token (from the request) for all requests that did not error, is this at all possible right now?

Given that I want to do this for all endpoints, using a middleware seems reasonable, but they have no way of accessing the route information (including the operation id). I can get the other bits of information from the requests/response no problem.

I also considered using a dependency, but even dependencies seem unable to have access to the operation id.

Operating System

macOS

Operating System Details

No response

FastAPI Version

0.65.2

Python Version

Python 3.9.1

Additional Context

No response

[dstlny]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options point_up_2

Example Code

@app.middleware("http")
async def log(request, call_next):
    response = call_next(request)
    if response.status_code >= 300:
        return response
    operation_id = get_operation_id_somehow() # is this possible?
    print("LOG", operation_id, get_access_token(request))
    return response

Description

I want to log the Operation ID, as well as the access token (from the request) for all requests that did not error, is this at all possible right now?

Given that I want to do this for all endpoints, using a middleware seems reasonable, but they have no way of accessing the route information (including the operation id). I can get the other bits of information from the requests/response no problem.

I also considered using a dependency, but even dependencies seem unable to have access to the operation id.

Operating System

macOS

Operating System Details

No response

FastAPI Version

0.65.2

Python Version

Python 3.9.1

Additional Context

No response

Something like the below would work:

@app.middleware("http")
async def log(request, call_next):
  response = await call_next(request)

  request_route = None

  for route in app.routes:
    if route.path == request.scope['path']:
      request_route = route
      break
      
  def get_operation_id_somehow():
    return request_route.unique_id or request_route.operation_id
  
  if request_route:
    operation_id = get_operation_id_somehow()
  
  print("LOG", operation_id)
  return response

@app.get('/', operation_id="some_id")
async def test(request):
	return {"msg": "Hello World"}

[ojii]

@dstlny thank you for the suggestion. I don't think that would work on variadic paths, would it? Also, I'm using sub-apps/sub-routers so would have to further change it to understand that. It also feels wrong to do route resolution twice (fastapi already does it once to run the actual request), is there really no better way than manually resolve the route again?

[dstlny]

@dstlny thank you for the suggestion. I don't think that would work on variadic paths, would it? Also, I'm using sub-apps/sub-routers so would have to further change it to understand that. It also feels wrong to do route resolution twice (fastapi already does it once to run the actual request), is there really no better way than manually resolve the route again?

I think it should work in most cases. I was under the impression that at some point FastAPI/Starlette included the route in which the path matched in the request object, but I appear to be wrong. I don't think there's much we can do here but resolve it twice to do what you wanna do. You might be able to pre-build this in a data-structure, say

PATH_LOOKUPS = { "route_path": "route_operation_id" }

after you've mounted all your sub-routers/sub-apps, so all you'd need to do is a simple X in PATH_LOOKUP or operation_id = PATH_LOOKUP[request.scope['path']] if performance is an issue.

[Kludex]

I can't read everything, so you can ignore if it's not useful, but: https://github.com/tomwojcik/starlette-context

[ojii]

Okay, this works but it does feel silly to do:

@app.middleware("http")
async def print_op_id(request, call_next):
    response = await call_next(request)
    print(request.method, request.url.path, op_id(app.routes, request.scope))
    return response


def op_id(routes, scope):
    for route in routes:
        if route.matches(scope):
            if isinstance(route, APIRoute):
                return route.operation_id
            elif isinstance(route, Mount):
                return op_id(route.routes, scope)

[panla]

from fastapi import Request
from fastapi.routing import APIRoute

class Route(APIRoute):
    """Extension Route and log request.json()"""

    def get_route_handler(self):
        original_route_handler = super().get_route_handler()

        async def log_request_some(request: Request):
            logger.info('some some some')
            return await original_route_handler(request)
        return log_request_detail

router = APIRouter(route_class=Route)

[ifigueroap]

I'm currently working on the same situation... this seems to work so far:

"""Instrumentation for registering requests and responses."""

import logging
from typing import Callable

from fastapi import Request, Response
from fastapi.routing import APIRoute

logger = logging.getLogger(__name__)


class LoggingRoute(APIRoute):  # noqa
    def get_route_handler(self) -> Callable:  # noqa
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:  # noqa
            logger.warning("====REQUEST SCOPE====")
            logger.warning(request.scope)
            logger.warning("=====================")

            logger.warning("====REQUEST BODY JSON====")
            body_json = await request.json()
            logger.warning(body_json)
            logger.warning("=====================")

            response = await original_route_handler(request)

            logger.warning("====RESPONSE JSON====")
            logger.warning(response.body)
            logger.warning("=====================")
            return response

        return custom_route_handler

However, you need to inject this class into the router before adding routes to it because the (internal?) add_route method uses the route_class constructor to create the routes, hence it cannot be overriden afterwards...

An example on how to do this:

router = APIRouter(prefix="/foo", route_class=LoggingRoute)

@router.post(...)
def post_method(...)
    # etc

[ojii]

I've ended up using the solution suggested in #3877 (comment)