207 lines
6.4 KiB
Python
207 lines
6.4 KiB
Python
import enum
|
|
|
|
from typing import Optional, Union, Dict, Any, List
|
|
from pydantic import BaseModel
|
|
from Controllers.Redis.database import RedisActions
|
|
|
|
|
|
class UserType(enum.Enum):
|
|
|
|
employee = 1
|
|
occupant = 2
|
|
|
|
|
|
class Credentials(BaseModel):
|
|
|
|
person_id: int
|
|
person_name: str
|
|
|
|
|
|
class ApplicationToken(BaseModel):
|
|
# Application Token Object -> is the main object for the user
|
|
|
|
user_type: int = UserType.occupant.value
|
|
credential_token: str = ""
|
|
|
|
user_uu_id: str
|
|
user_id: int
|
|
|
|
person_id: int
|
|
person_uu_id: str
|
|
|
|
request: Optional[dict] = None # Request Info of Client
|
|
expires_at: Optional[float] = None # Expiry timestamp
|
|
|
|
|
|
class OccupantToken(BaseModel):
|
|
|
|
# Selection of the occupant type for a build part is made by the user
|
|
|
|
living_space_id: int # Internal use
|
|
living_space_uu_id: str # Outer use
|
|
|
|
occupant_type_id: int
|
|
occupant_type_uu_id: str
|
|
occupant_type: str
|
|
|
|
build_id: int
|
|
build_uuid: str
|
|
build_part_id: int
|
|
build_part_uuid: str
|
|
|
|
responsible_company_id: Optional[int] = None
|
|
responsible_company_uuid: Optional[str] = None
|
|
responsible_employee_id: Optional[int] = None
|
|
responsible_employee_uuid: Optional[str] = None
|
|
|
|
# ID list of reachable event codes as "endpoint_code": ["UUID", "UUID"]
|
|
reachable_event_codes: Optional[dict[str, list[str]]] = None
|
|
|
|
# ID list of reachable applications as "page_url": ["UUID", "UUID"]
|
|
reachable_app_codes: Optional[dict[str, list[str]]] = None
|
|
|
|
|
|
class CompanyToken(BaseModel):
|
|
|
|
# Selection of the company for an employee is made by the user
|
|
company_id: int
|
|
company_uu_id: str
|
|
|
|
department_id: int # ID list of departments
|
|
department_uu_id: str # ID list of departments
|
|
|
|
duty_id: int
|
|
duty_uu_id: str
|
|
|
|
staff_id: int
|
|
staff_uu_id: str
|
|
|
|
employee_id: int
|
|
employee_uu_id: str
|
|
bulk_duties_id: int
|
|
|
|
# ID list of reachable event codes as "endpoint_code": ["UUID", "UUID"]
|
|
reachable_event_codes: Optional[dict[str, list[str]]] = None
|
|
|
|
# ID list of reachable applications as "page_url": ["UUID", "UUID"]
|
|
reachable_app_codes: Optional[dict[str, list[str]]] = None
|
|
|
|
|
|
class OccupantTokenObject(ApplicationToken):
|
|
# Occupant Token Object -> Requires selection of the occupant type for a specific build part
|
|
|
|
available_occupants: dict = None
|
|
selected_occupant: Optional[OccupantToken] = None # Selected Occupant Type
|
|
|
|
@property
|
|
def is_employee(self) -> bool:
|
|
return False
|
|
|
|
@property
|
|
def is_occupant(self) -> bool:
|
|
return True
|
|
|
|
|
|
class EmployeeTokenObject(ApplicationToken):
|
|
# Full hierarchy Employee[staff_id] -> Staff -> Duty -> Department -> Company
|
|
|
|
companies_id_list: List[int] # List of company objects
|
|
companies_uu_id_list: List[str] # List of company objects
|
|
|
|
duty_id_list: List[int] # List of duty objects
|
|
duty_uu_id_list: List[str] # List of duty objects
|
|
|
|
selected_company: Optional[CompanyToken] = None # Selected Company Object
|
|
|
|
@property
|
|
def is_employee(self) -> bool:
|
|
return True
|
|
|
|
@property
|
|
def is_occupant(self) -> bool:
|
|
return False
|
|
|
|
|
|
TokenDictType = Union[EmployeeTokenObject, OccupantTokenObject]
|
|
|
|
|
|
class TokenProvider:
|
|
|
|
AUTH_TOKEN: str = "AUTH_TOKEN"
|
|
|
|
@classmethod
|
|
def convert_redis_object_to_token(
|
|
cls, redis_object: Dict[str, Any]
|
|
) -> TokenDictType:
|
|
"""
|
|
Process Redis object and return appropriate token object.
|
|
"""
|
|
if redis_object.get("user_type") == UserType.employee.value:
|
|
return EmployeeTokenObject(**redis_object)
|
|
elif redis_object.get("user_type") == UserType.occupant.value:
|
|
return OccupantTokenObject(**redis_object)
|
|
raise ValueError("Invalid user type")
|
|
|
|
@classmethod
|
|
def get_dict_from_redis(
|
|
cls, token: Optional[str] = None, user_uu_id: Optional[str] = None
|
|
) -> Union[TokenDictType, List[TokenDictType]]:
|
|
"""
|
|
Retrieve token object from Redis using token and user_uu_id
|
|
"""
|
|
token_to_use, user_uu_id_to_use = token or "*", user_uu_id or "*"
|
|
list_of_token_dict, auth_key_list = [], [
|
|
cls.AUTH_TOKEN,
|
|
token_to_use,
|
|
user_uu_id_to_use,
|
|
]
|
|
if token:
|
|
result = RedisActions.get_json(list_keys=auth_key_list, limit=1)
|
|
if first_record := result.first:
|
|
return cls.convert_redis_object_to_token(first_record)
|
|
elif user_uu_id:
|
|
result = RedisActions.get_json(list_keys=auth_key_list)
|
|
if all_records := result.all:
|
|
for all_record in all_records:
|
|
list_of_token_dict.append(
|
|
cls.convert_redis_object_to_token(all_record)
|
|
)
|
|
return list_of_token_dict
|
|
raise ValueError(
|
|
"Token not found in Redis. Please check the token or user_uu_id."
|
|
)
|
|
|
|
@classmethod
|
|
def retrieve_application_codes(
|
|
cls, page_url: str, tokens: Union[TokenDictType, List[TokenDictType]]
|
|
):
|
|
if isinstance(tokens, TokenDictType):
|
|
if application_codes := tokens.reachable_app_codes.get(page_url, None):
|
|
return application_codes
|
|
elif isinstance(tokens, list):
|
|
retrieved_event_apps = []
|
|
for token in tokens:
|
|
if not isinstance(token, TokenDictType):
|
|
continue
|
|
if application_codes := token.reachable_app_codes.get(page_url, None):
|
|
retrieved_event_apps.append(application_codes)
|
|
return retrieved_event_apps
|
|
raise ValueError("Invalid token type or no application codes found.")
|
|
|
|
@classmethod
|
|
def retrieve_event_codes(
|
|
cls, endpoint_code: str, tokens: Union[TokenDictType, List[TokenDictType]]
|
|
):
|
|
if isinstance(tokens, TokenDictType):
|
|
if event_codes := tokens.reachable_event_codes.get(endpoint_code, None):
|
|
return event_codes
|
|
elif isinstance(tokens, List):
|
|
retrieved_event_codes = []
|
|
for token in tokens:
|
|
if not isinstance(token, TokenDictType):
|
|
continue
|
|
if event_codes := token.reachable_event_codes.get(endpoint_code, None):
|
|
retrieved_event_codes.append(event_codes)
|
|
return retrieved_event_codes
|
|
raise ValueError("Invalid token type or no event codes found.")
|