How does async __call__ work in HTTPBearer and where is it invoked? #13336
-
First Check
Commit to Help
Example Codefrom typing import Annotated
from fastapi import Depends
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
class ValidateHTTPBearer(HTTPBearer):
async def validate_parse_token(self, token: str):
...
async def validate_current_user(self, parsed_token: dict):
...
async def __call__(self, request: Request):
...
parsed_token = await self.validate_parse_token("token")
await self.validate_current_user(parsed_token)
...
WhoAreYou = Depends(ValidateHTTPBearer())
CurrentUser = Annotated[HTTPAuthorizationCredentials, WhoAreYou]DescriptionI have seen HTTPBearer class in fastapi and it uses Async call method and then use dependency injection. I have created a sub-class using HTTPBearer having additional validation logic at dependency level. Do we invoke in call that inherits async class like
If token is valid check then validate user Operating SystemLinux Operating System DetailsFedora 41 FastAPI Version0.115.7 Pydantic Version2.10 Python Version3.12 Additional ContextNo response |
Beta Was this translation helpful? Give feedback.
Replies: 1 comment · 2 replies
-
|
The answer for the first question: FastAPI analyzes parameters of the endpoint function and executes I think it's a mistake in Regarding your second question - I didn't get it. Could you clarify? Do you want to achieve something like this? |
Beta Was this translation helpful? Give feedback.
-
|
My Bad. It should have been If I need to create similar functionality using parent and sub-class. where should I invoke parent Regarding the second question. I need to write custom class which inherits HTTPBearer and write custom validation. Instead if HTTPBearer accepts another optional parameter say Currently, HTTPBearer accepts four parameters HTTPBearer(schema_name="SomeBearer", other_validations=[validate_token(), validate_user()]) |
Beta Was this translation helpful? Give feedback.
-
|
So, as I understand, this is what you needed: from collections.abc import Awaitable, Callable
from typing import Annotated, Optional
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.security import HTTPBearer
class ValidateHTTPBearer(HTTPBearer):
def __init__(
self,
*,
scheme_name: Optional[str] = None,
other_validations: list[Callable[[str], Awaitable[None]]],
):
super().__init__(scheme_name=scheme_name)
self._other_validations = other_validations
async def validate_parse_token(self, token: str) -> dict:
data = token.split(":")
return {"user": data[0], "something": data[1]}
async def validate_current_user(self, parsed_token: dict) -> str:
return parsed_token["user"]
async def __call__(self, request: Request) -> str:
# Call the parent class to get the credentials
auth_credentials = await super().__call__(request)
# Validate the token and user
parsed_token = await self.validate_parse_token(auth_credentials.credentials)
user = await self.validate_current_user(parsed_token)
# Perform other validations
token_data = parsed_token.copy()
token_data["user"] = user
for validation in self._other_validations:
await validation(token_data)
return user
async def validate_token(token_data: dict):
if token_data["something"] != "123":
raise HTTPException(status_code=401, detail="Invalid token")
async def validate_user(token_data: dict):
if token_data.get("user") != "admin":
raise HTTPException(status_code=403, detail="Invalid user")
security = ValidateHTTPBearer(
scheme_name="SomeBearer",
other_validations=[validate_token, validate_user]
)
CurrentUser = Annotated[str, Depends(security)]
app = FastAPI()
@app.get("/")
def index(current_user: CurrentUser):
return {"user": current_user}And the answer to your question:
FastAPI analyzes dependencies of the endpoint and calls\awaits them depending on the function signature. |
Beta Was this translation helpful? Give feedback.
So, as I understand, this is what you needed: