Programmatically wrap all routes in APIRouter in custom decorator(s)

[richardnm-2]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question 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

import random
import time
import warnings

import uvicorn
from fastapi import APIRouter, FastAPI
from fastapi.dependencies.utils import get_dependant
from pydantic import BaseModel

app = FastAPI()

router = APIRouter()


class Alert(BaseModel):
    message: str
    alert_type: str


def deep_stacked_func():
    '''
        Might be called by other functions, be called outside a route and be declared on a different package.
        Don't want to have to receive a request object to have the session that the call belongs.
        '''
    n = random.randint(1, 10)
    if n <= 8:
        # do function stuff

        alert = Alert(message='/warnings route', alert_type='error')
        warnings.warn(Warning(alert))

    return 'function stuff ' * n   # must be returned, even after warnings.warn call


def alerts_decorator(func):
    async def wrapper():
        with warnings.catch_warnings(record=True) as caught_warnings:
            warnings.simplefilter("always")
            response = await func()

            warns = []

            for warn in caught_warnings:
                warn_message = warn.message.args[0]

                if isinstance(warn_message, Alert):
                    warns.append(
                        {'message': warn_message.message, 'type': warn_message.alert_type})

            if warns:
                response.update({'alerts': warns})
            return response

    return wrapper


@router.get("/warnings")
async def warnings_route():

    function_stuff = deep_stacked_func()

    # do route stuff, with function stuff

    time.sleep(random.random())

    return {"return": "route stuff"}


@router.get("/no_warnings")
async def no_warnings_route():
    time.sleep(random.random())
    return {"return": "no warning route stuff"}


for route in router.routes:
    route.endpoint = alerts_decorator(route.endpoint)
    route.dependendant = get_dependant(
        path=route.path_format, call=route.endpoint)

app.include_router(router)

if __name__ == '__main__':
    uvicorn.run('app:app', host='0.0.0.0', port=8000, reload=True)

Description

After an issue with Starlette BaseHTTPMiddleware catching warnings using the warnings package, questioned here, I tried to achieve the same end result (appending warnings caught within a warnings.catch_warnings context) wrapping the endpoint in a custom decorator, and it worked (even got simpler without having to read the contents of response.body_iterator, update the body and create a new Response instance).

I just did not want to go back to my API and wrap every route endpoint function in this new decorator, nor I'd like to have me or anyone else to remember to do so in new routes.

I achieved the expected behavior by looping through all routes inside the router and editing the endpoint property of the route, as well as the dependant property:

from fastapi.dependencies.utils import get_dependant

router = APIRouter()

...

for route in router.routes:
    route.endpoint = alerts_decorator(route.endpoint)
    route.dependendant = get_dependant(
        path=route.path_format, call=route.endpoint)

It would be nice to be able to do so in the router instantiation, as with prefix, or some argument on the inclusion of the router in the app

Operating System

Windows

Operating System Details

No response

FastAPI Version

0.94.0

Python Version

3.10.7

Additional Context

No response

[tiangolo]

Hey there!

You shouldn't try to use warnings.catch_warnings() in anything that expects to run more than one thing during the same period of time, like a web API handling requests. It is not thread-safe, so it's not concurrent-safe, it's only for smaller or simpler scripts, during testing or so. You can read more in the official docs for warnings, in the last note: https://docs.python.org/3/library/warnings.html#available-context-managers

For what you are doing, if you want to capture warnings per-request in a similar way, you would have to create your own utils equivalent to:

• Create a context var to store the warnings per request
• Set the context var to a list or similar in the middleware or a dependency with yield
• Create a utility function equivalent to warnings.warn() that would store the warning in that list in the context var
• Use the contents of the context var list in the way you want
• After that, clean up the context var