74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
import json
|
|
import typing
|
|
|
|
from .conn import redis_cli
|
|
|
|
from api_configs import Auth
|
|
from api_objects import EmployeeTokenObject, OccupantTokenObject
|
|
|
|
|
|
def get_object_via_access_key(
|
|
request,
|
|
) -> typing.Union[EmployeeTokenObject, OccupantTokenObject, None]:
|
|
from fastapi import status
|
|
from fastapi.exceptions import HTTPException
|
|
|
|
if not hasattr(request, "headers"):
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail=dict(
|
|
message="Headers are not found in request. Invalid request object."
|
|
),
|
|
)
|
|
if not request.headers.get(Auth.ACCESS_TOKEN_TAG):
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail=dict(message="Unauthorized user, please login..."),
|
|
)
|
|
already_tokens = redis_cli.scan_iter(
|
|
match=str(request.headers.get(Auth.ACCESS_TOKEN_TAG) + ":*")
|
|
)
|
|
if already_tokens := list(already_tokens):
|
|
try:
|
|
if redis_object := json.loads(
|
|
redis_cli.get(already_tokens[0].decode()) or {}
|
|
):
|
|
if redis_object.get("user_type") == 1:
|
|
if not redis_object.get("selected_company", None):
|
|
redis_object["selected_company"] = None
|
|
return EmployeeTokenObject(**redis_object)
|
|
elif redis_object.get("user_type") == 2:
|
|
if not redis_object.get("selected_occupant", None):
|
|
redis_object["selected_occupant"] = None
|
|
return OccupantTokenObject(**redis_object)
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail=dict(
|
|
message="User type is not found in the token object. Please reach to your administrator."
|
|
),
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={
|
|
"message": "Redis Service raised an exception.",
|
|
"error": str(e),
|
|
},
|
|
)
|
|
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid credentials. Please login again.",
|
|
)
|
|
|
|
|
|
def get_object_via_user_uu_id(user_id: str) -> typing.Union[dict, None]:
|
|
already_tokens = redis_cli.scan_iter(match=str("*:" + str(user_id)))
|
|
already_tokens_list, already_tokens_dict = [], {}
|
|
for already_token in already_tokens:
|
|
redis_object = json.loads(redis_cli.get(already_token) or {})
|
|
already_tokens_list.append(redis_object)
|
|
already_tokens_dict[already_token.decode()] = redis_object
|
|
return already_tokens_dict
|
|
|