As the title suggests, I want to use FastAPI’s Depends to implement role-based authentication control for specific routes.
I have two files. If I write the authentication logic as shown below:
auth.py file
class Authenticator:
async def get_sub(self, token: Annotated[str, Depends(oauth2_scheme)]) -> str:
** asynchronous operation **
return sub
async def get_user(self, sub: Annotated[User, Depends(get_sub)]) -> UserInDB:
sub = await self.get_sub(token)
user = await self.user_db.get_user_by_id(id=id)
if not user:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not register yet")
return user
authenticator = Authenticator()
router.py file
router = APIRouter(prefix="/user", dependencies=[Depends(authenticator.get_user)])
@router.get("/")
async def protected_route(request: Request):
pass
If I write the get_user function in the first file with sub: Annotated[User, Depends(get_sub)], a “self” query parameter appears in the route when I run it, which seems odd. How can I make this asynchronous Depends chain work properly without the “self” issue?
know how asynchronous Depends chain work
2
If you don’t need to use self in get_sub, you can make get_sub static method:
class Authenticator:
@staticmethod
async def get_sub(token: Annotated[str, Depends(oauth2_scheme)]) -> str:
** asynchronous operation **
return sub
async def get_user(self, sub: Annotated[User, Depends(get_sub)]) -> UserInDB:
# sub = await self.get_sub(token)
user = await self.user_db.get_user_by_id(id=id)
if not user:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not register yet")
return user
authenticator = Authenticator()
Another way is to implement this without class:
db = UserDatabase()
async def get_user_db():
yield db
async def get_sub(token: Annotated[str, Depends(oauth2_scheme)]):
#
return sub
async def get_user(
sub: Annotated[User, Depends(get_sub)],
user_db: Annotated[UserDatabase, Depends(get_user_db)],
) -> UserInDB:
user = await user.get_user_by_id(id=id)
if not user:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not register yet")
return user
app = FastAPI()
@app.get("/")
def index(user: Annotated[UserInDB, Depends(get_user)]):
pass
3