How to add a custom decorator to a FastAPI route?

Question

I want to add an auth_required decorator to my endpoints. (Please consider that this question is about decorators, not middleware)

So a simple decorator looks like this:

def auth_required(func):
 def wrapper(*args, **kwargs):
 if user_ctx.get() is None:
 raise HTTPException(...)
 return func(*args, **kwargs)
 return wrapper

So there are 2 usages:

@auth_required
@router.post(...)

or

@router.post(...)
@auth_required

The first way doesn't work because router.post creates a router that saved into self.routes of APIRouter object. The second way doesn't work because it fails to verify pydantic object. For any request model, it says missing args, missing kwargs.

So my question is - how can I add any decorators to FastAPI endpoints? Should I get into router.routes and modify the existing endpoint? Or use some functools.wraps like functions?

• Gino Mempin: Is there a reason you need it to be a decorator? Coming from Flask to FastAPI, I sometimes think I need a decorator, but a custom APIRoute class for endpoints that need auth or a Depends(User) injection can also solve the problem.
• sashaaero: I want to add that decorator to some endpoints, not every. So custom APIRoute class (Im actually using it) doesnt help. And I have an issue with middleware - it works in another thread, so I can't set up global context variable from another thread. I saw some solutions to it, but now i really want to know is decorators possible.
• Gino Mempin: The recommended style with FastAPI seems to be to use Dependencies. You add something like user: User = Depends(auth_function) to the path or function. That gets called before your endpoint function, similar to how a decorator wraps it. It should also have access to the req-resp context.
• sashaaero: I know how to use depends. It has access to context, but since it is working in another thread, im getting empty context in main thread.
• Shannon: "Depends" also can't do any "around" actions (it can't do stuff after the route method body has completed.

Answer 1

How can I add any decorators to FastAPI endpoints?

As you said, you need to use @functools.wraps(...)--(PyDoc) decorator as,

from functools import wraps
from fastapi import FastAPI
from pydantic import BaseModel
class SampleModel(BaseModel):
 name: str
 age: int
app = FastAPI()
def auth_required(func):
 @wraps(func)
 async def wrapper(*args, **kwargs):
 return await func(*args, **kwargs)
 return wrapper
@app.post("/")
@auth_required # Custom decorator
async def root(payload: SampleModel):
 return {"message": "Hello World", "payload": payload}

The main caveat of this method is that you can't access the request object in the wrapper and I assume it is your primary intention.

If you need to access the request, you must add the argument to the router function as,

from fastapi import Request
@app.post("/")
@auth_required # Custom decorator
async def root(request: Request, payload: SampleModel):
 return {"message": "Hello World", "payload": payload}

---

I am not sure what's wrong with the FastAPI middleware, after all, the @app.middleware(...) is also a decorator.

• uberrebu: can you please elaborate on the @app.middleware(...) you mean that can work as decorators also? any example or tutorial of this?
• Ankit Jain: This works for post requests but not for get requests. Any idea why?
• JPG: @auth_required is independent of the request method.
• Fahad Munir: @AnkitJain Please make sure to declare the endpoint functions with async def. Above @auth_required decorator will only work with functions declared with async def. If you don't use async/await in endpoint functions, just drop it from def wrapper(*args, **kwargs) definition.
• Ankit Jain: @FahadMunir Thanks a lot! This worked. I have added async in endpoint functions. However, this was not an issue with post requests.

Answer 2

Simply use the dependencies inside of the path operation decorator:

from fastapi import Depends, FastAPI, Header, HTTPException
app = FastAPI()
async def verify_token(x_token: str = Header()):
 if x_token != "fake-super-secret-token":
 raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: str = Header()):
 if x_key != "fake-super-secret-key":
 raise HTTPException(status_code=400, detail="X-Key header invalid")
 return x_key
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
 return [{"item": "Foo"}, {"item": "Bar"}]

• Cyril Gupta: This is not a custom decorator though.

Answer 3

Here is how you can use a decorator that adds extra parameters to the route handler:

from fastapi import FastAPI, Request
from pydantic import BaseModel
class SampleModel(BaseModel):
 name: str
 age: int
app = FastAPI()
def do_something_with_request_object(request: Request):
 print(request)
def auth_required(handler):
 async def wrapper(request: Request, *args, **kwargs):
 do_something_with_request_object(request)
 return await handler(*args, **kwargs)
 # Fix signature of wrapper
 import inspect
 wrapper.__signature__ = inspect.Signature(
 parameters = [
 # Use all parameters from handler
 *inspect.signature(handler).parameters.values(),
 # Skip *args and **kwargs from wrapper parameters:
 *filter(
 lambda p: p.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD),
 inspect.signature(wrapper).parameters.values()
 )
 ],
 return_annotation = inspect.signature(handler).return_annotation,
 )
 return wrapper
@app.post("/")
@auth_required # Custom decorator
async def root(payload: SampleModel):
 return {"message": f"Hello {payload.name}, {payload.age} years old!"}

• JPG: Tried, but got TypeError: wrapper() missing 1 required positional argument: 'request'
• md2perpe: @JPG. I have updated the code after a lot of investigation and testing. However, I have tested with my own code; I haven't tested the code above.
• md2perpe: gist.github.com/md2perpe/ee146e547a0bd910ea9683a2eea47c59
• gocreating: same error TypeError: wrapper() missing 1 required positional argument: 'request'
• gocreating: I have answered myself on fastapi's issue: github.com/tiangolo/fastapi/issues/2662

Answer 4

In addtion to JPG's answer, you can access the Request object inside your decorator with kwargs.get('request'). A full decorator would look something like:

def render_template(template):
 """decorator to render a template with a context"""
 def decorator(func):
 @wraps(func)
 def wrapper(*args, **kwargs):
 # access request object
 request = kwargs.get('request')
 context = func(*args, **kwargs)
 if context is None:
 context = {}
 return templates.TemplateResponse(template, {**context, 'request': request})
 return wrapper
 return decorator

The decorated function will need to take the Request as a parameter, however.

Answer 5

The accepted answer won't work if your decorator needs to access FastAPI dependencies (e.g. request info, auth schemes, database sessions, ...).

If you need your endpoint decorators to use FastAPI dependencies, you can use the @depends() decorator from fastapi-decorators to achieve it:

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def authorize(scope: str):
 @depends
 def decorator(token: str = Depends(oauth2_scheme)):
 jwt = jwt_decode(token)
 if scope in not jwt.scopes:
 raise HTTPException(status_code=403, detail="Unauthorized")
 return decorator
@app.put("/users/{user_id}")
@authorize("users:write")
def update_user(*, user_id: int, user_update: UserUpdate):
 ...

Answer 6

The proposed solution can depending on the situation lead to problems. In my case, the decorator was not an "auth_required" but a "use_cached" answer one and I was using a nonlocal var. As a result I was getting:

RuntimeError: cannot reuse already awaited coroutine

This is happening because as the code is proposed, it uses async / await in two locations / functions. If you get this error, to avoid it, you need to get rid of the extra async / await.

This has the disadvantage of breaking the aesthetics and that adding your decorator slightly changes the following function (you drop async). It has the advantage that the code produces no errors and is correct however...

TL;DR:

def auth_required(func):
 @wraps(func)
 async def wrapper(*args, **kwargs):
 return func(*args, **kwargs) #DO NOT WAIT
 return wrapper
@app.post("/")
@auth_required # Custom decorator
def root(payload: SampleModel): #NOT ASYNC
 return {"message": "Hello World", "payload": payload}