Closed techxonia closed 3 years ago
You return error, not return ex. check your code again when I testing in my machine is working
@app.get('/user')
def user(Authorize: AuthJWT = Depends()):
try:
current_user = jwt_token_checker(Authorize)
except Exception as ex:
return ex # return ex not error
return {"user": current_user}
def jwt_token_checker(Authorize: AuthJWT):
try:
Authorize.jwt_required()
return Authorize.get_jwt_subject()
except Exception as ex:
return ex # return ex not error
Here is actual code: It is not working:
@router.get('/user')
async def items(Authorize: AuthJWT = Depends(), db: AsyncIOMotorClient = Depends(get_database)):
try:
""" If there was an exception, the function must throw an error and must not execute next line of code """"
await jwt_token_checker(Authorize, db)
""" but now it always returns a success true if wrong token pass """
return {"success": true}
except Exception as ex:
return JSONResponse(
status_code=200,
content={
"data": None,
"code": 400,
"message": "Some thing Went Wrong,
"status": False
}
)
Helper Function:
async def jwt_token_checker(language: str, Authorize: AuthJWT , conn: AsyncIOMotorClient):
try:
Authorize.jwt_required()
jti = Authorize.get_raw_jwt()['jti']
current_user = Authorize.get_jwt_subject()
access_token_check = await check_access_token(conn, Authorize, jti, current_user)
errorMessage = ""
if access_token_check == AuthEnum.FORBIDDEN_UNAUTHORIZED_ACCESS:
errorMessage = Locale(
language, strings.FORBIDDEN_UNAUTHORIZED_ACCESS).string
return JSONResponse(
status_code=200,
content={
"data": None,
"code": 400,
"message": errorMessage
"status": False
}
)
except AuthJWTException as e:
return JSONResponse(
status_code=401,
content={
"data": None,
"code": 401,
"message": "UnAuthorize User"
"status": False
}
)
you must raise an exception not return a response
@app.get('/user')
async def user(Authorize: AuthJWT = Depends()):
try:
await jwt_token_checker(Authorize)
return {"success": True}
except Exception:
return JSONResponse(
status_code=200,
content={
"data": None,
"code": 400,
"message": "Some thing Went Wrong",
"status": False
}
)
async def jwt_token_checker(Authorize: AuthJWT):
try:
Authorize.jwt_required()
except AuthJWTException:
raise # raise exception
I try to separate the logic of checking JWT tokens from one function so that it can be reused. But when I try to run the code, it does not work. I manage an access token in my database and check if it exists or not. The last generated token will be valid only for the access token. This access token must be checked at every protected endpoint. Here is my code:
Helper Function:
Is this right way to separate the token checker function? If not then can you please suggest me some solution.