I want to decorate my path operation function using a custom decorator. I want to check for some information within the Request body for every request and based on the information, Permit or disallow the user from accessing certain endpoints.
I'm wondering if this can be done, And if so, How? I tried a few options but I'm not able to get it working as expected.
This is my decorator function:
def decorator(permissions):
def wrapper(function, *args, **kwargs):
async def executor():
return await function(*args, **kwargs)
return executor
return wrapper
And this is how I am using it
@apiRouter.get("/example-end-point")
@decorator(["CAN_DO_SOMETHING"])
async def do_something():
"""
Do whatever needs to be done
"""
return {"payload": {}}
I am able to decorate the function in the sense that the do_something function executes every time the end-point is called, along with decorator code. BUT, I face problems such as
async def do_something(query: str)
Returns an ASGI Internal Server Error instead of a 422 / Validation Error; EVEN when I pass the query parameter.
I am assuming that the query parameters are not "passed" to the end-point function triggering this error, but I'm curious as to why FastAPI does not handle it with a 422, instead throwing an ASGI Exception with a TypeError as
TypeError: do_something() missing 1 required positional argument: 'query'
WARNING: Executing <Task finished coro=<RequestResponseCycle.run_asgi() done, defined at /project/env/lib/python3.7/site-packages/uvicorn/protocols/http/httptools_impl.py:373> result=None created at /project/env/lib/python3.7/site-packages/uvicorn/protocols/http/httptools_impl.py:261> took 3.402 seconds
which I assume occurs when a finished co-routine object is awaited.
I tried searching online but could not find much regarding fastapi custom decorators OR the warning <Task finished...done>.
Would appreciate any advice I get.
Thanks a bunch!
Try using built-in functools.wraps() that preserves function signature. In your case I would rewrite code like that:
from functools import wraps
def decorator(permissions):
def outer_wrapper(function):
@wraps(function)
async def inner_wrapper(*args, **kwargs):
return await function(*args, **kwargs)
return inner_wrapper
return outer_wrapper
Try and let me know if it helps.
@HarrySky Using functools.wrap() solves my problem completely! Thanks a bunch!
Also, Would you have any idea why using functools solves my problem? From what I understand, using functools preserves my function signature, But would the function signature be something that is depended on by FastAPI?
@subritc as far as I know, FastAPI checks endpoint function signature on startup so that it knows what kind of data should be provided to the endpoint, that's why it was not providing query: str argument - it simply did not detect it :smile:
Got it. Without the functools, FastAPI was throwing a 422 for parameters args and *kwargs which was the signature of the wrapping decorator function, But using functools preserves the signature for my endpoint function.
Thanks a lot @HarrySky :)
Another great way of doing such kind of things, is using the amazing depends. You can create an simple function and tell that a request depends on that function execution, that is, doing checks or all that you need before executing inner_wrap.
PS: you call use depends beautifully by:
permissions = Depends(auth)
@router.get(
"/path",
depends=[permissions]
)
Thanks for the help here @HarrySky! :clap: :bow:
And yeah, all what @HarrySky said :point_up_2:
Thanks for reporting back and closing the issue @subritc :+1:
Also, as @gabriel-viviani says, depends would be the best way to go to handle shared logic like permissions.
Most helpful comment
@subritc as far as I know, FastAPI checks endpoint function signature on startup so that it knows what kind of data should be provided to the endpoint, that's why it was not providing
query: strargument - it simply did not detect it :smile: