355 lines
13 KiB
Python
355 lines
13 KiB
Python
"""Token service for handling authentication tokens and user sessions."""
|
|
|
|
import json
|
|
from typing import List, Union, TypeVar, Dict, Any, Optional, TYPE_CHECKING
|
|
|
|
from AllConfigs.Token.config import Auth
|
|
from ApiLibrary.common.line_number import get_line_number_for_error
|
|
from ApiLibrary.date_time_actions.date_functions import DateTimeLocal
|
|
from ApiLibrary.token.password_module import PasswordModule
|
|
from ErrorHandlers import HTTPExceptionApi
|
|
from Schemas.identity.identity import UsersTokens
|
|
from Services.Redis import RedisActions, AccessToken
|
|
from ApiValidations.Custom.token_objects import (
|
|
EmployeeTokenObject,
|
|
OccupantTokenObject,
|
|
UserType,
|
|
)
|
|
from Schemas import (
|
|
Users,
|
|
BuildLivingSpace,
|
|
BuildParts,
|
|
Employees,
|
|
Addresses,
|
|
Companies,
|
|
Staff,
|
|
Duty,
|
|
Duties,
|
|
Departments,
|
|
OccupantTypes,
|
|
)
|
|
from Services.Redis.Models.base import RedisRow
|
|
from Services.Redis.Models.response import RedisResponse
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import Request
|
|
|
|
T = TypeVar("T", EmployeeTokenObject, OccupantTokenObject)
|
|
|
|
|
|
class TokenService:
|
|
"""Service class for handling authentication tokens and user sessions."""
|
|
|
|
@classmethod
|
|
def _create_access_token(cls, access: bool = True) -> str:
|
|
"""Generate a new access token."""
|
|
if not access:
|
|
return PasswordModule.generate_token(Auth.REFRESHER_TOKEN_LENGTH)
|
|
return PasswordModule.generate_token(Auth.ACCESS_TOKEN_LENGTH)
|
|
|
|
@classmethod
|
|
def _get_user_tokens(cls, user: Users) -> RedisResponse:
|
|
"""Get all tokens for a user from Redis."""
|
|
return RedisActions.get_json(
|
|
list_keys=AccessToken(
|
|
userUUID=user.uu_id,
|
|
).to_list()
|
|
)
|
|
|
|
@classmethod
|
|
def do_occupant_login(
|
|
cls, request: "Request", user: Users, domain: str
|
|
) -> Dict[str, Any]:
|
|
"""Handle occupant login process and return login information."""
|
|
db_session = BuildLivingSpace.new_session()
|
|
living_spaces: list[BuildLivingSpace] = BuildLivingSpace.filter_all(
|
|
BuildLivingSpace.person_id == user.person_id, db=db_session
|
|
).data
|
|
if not living_spaces:
|
|
raise HTTPExceptionApi(
|
|
error_code="", lang="en", loc=get_line_number_for_error()
|
|
)
|
|
|
|
occupants_selection_dict: Dict[str, Any] = {}
|
|
|
|
for living_space in living_spaces:
|
|
build_parts_selection = BuildParts.filter_all(
|
|
BuildParts.id == living_space.build_parts_id,
|
|
db=db_session,
|
|
).data
|
|
if not build_parts_selection:
|
|
raise HTTPExceptionApi(
|
|
error_code="", lang="en", loc=get_line_number_for_error()
|
|
)
|
|
|
|
build_part = build_parts_selection.get(1)
|
|
build = build_part.buildings
|
|
occupant_type = OccupantTypes.filter_by_one(
|
|
id=living_space.occupant_type,
|
|
db=db_session,
|
|
system=True,
|
|
).data
|
|
|
|
occupant_data = {
|
|
"part_uu_id": str(build_part.uu_id),
|
|
"part_name": build_part.part_name,
|
|
"part_level": build_part.part_level,
|
|
"uu_id": str(occupant_type.uu_id),
|
|
"description": occupant_type.occupant_description,
|
|
"code": occupant_type.occupant_code,
|
|
}
|
|
|
|
build_key = str(build.uu_id)
|
|
if build_key not in occupants_selection_dict:
|
|
occupants_selection_dict[build_key] = {
|
|
"build_uu_id": build_key,
|
|
"build_name": build.build_name,
|
|
"build_no": build.build_no,
|
|
"occupants": [occupant_data],
|
|
}
|
|
else:
|
|
occupants_selection_dict[build_key]["occupants"].append(occupant_data)
|
|
model_value = OccupantTokenObject(
|
|
domain=domain,
|
|
user_type=UserType.occupant.value,
|
|
user_uu_id=str(user.uu_id),
|
|
credentials=user.credentials(),
|
|
user_id=user.id,
|
|
person_id=user.person_id,
|
|
person_uu_id=str(user.person.uu_id),
|
|
request=dict(request.headers),
|
|
available_occupants=occupants_selection_dict,
|
|
timezone=user.local_timezone or "GMT+0",
|
|
lang=user.lang or "tr",
|
|
).model_dump()
|
|
cls.set_object_to_redis(user, model_value)
|
|
return {
|
|
"user_type": UserType.occupant.name,
|
|
"available_occupants": occupants_selection_dict,
|
|
}
|
|
|
|
@classmethod
|
|
def set_object_to_redis(cls, user, model: Dict):
|
|
accessObject = AccessToken(
|
|
userUUID=user.uu_id,
|
|
accessToken=cls._create_access_token(),
|
|
)
|
|
return RedisActions.set_json(
|
|
list_keys=accessObject.to_list(),
|
|
value=json.dumps(model),
|
|
expires=Auth.TOKEN_EXPIRE_MINUTES_30.seconds,
|
|
)
|
|
|
|
@classmethod
|
|
def do_employee_login(
|
|
cls, request: "Request", user: Users, domain: str
|
|
) -> Dict[str, Any]:
|
|
"""Handle employee login process and return login information."""
|
|
db_session = Employees.new_session()
|
|
list_employee = Employees.filter_all(
|
|
Employees.people_id == user.person_id, db=db_session
|
|
).data
|
|
|
|
companies_uu_id_list: List[str] = []
|
|
companies_id_list: List[int] = []
|
|
companies_list: List[Dict[str, Any]] = []
|
|
duty_uu_id_list: List[str] = []
|
|
duty_id_list: List[int] = []
|
|
|
|
for employee in list_employee:
|
|
staff = Staff.filter_one(Staff.id == employee.staff_id, db=db_session).data
|
|
if duties := Duties.filter_one(
|
|
Duties.id == staff.duties_id, db=db_session
|
|
).data:
|
|
if duty_found := Duty.filter_by_one(
|
|
id=duties.duties_id, db=db_session
|
|
).data:
|
|
duty_uu_id_list.append(str(duty_found.uu_id))
|
|
duty_id_list.append(duty_found.id)
|
|
|
|
department = Departments.filter_one(
|
|
Departments.id == duties.department_id, db=db_session
|
|
).data
|
|
|
|
if company := Companies.filter_one(
|
|
Companies.id == department.company_id, db=db_session
|
|
).data:
|
|
companies_uu_id_list.append(str(company.uu_id))
|
|
companies_id_list.append(company.id)
|
|
company_address = Addresses.filter_by_one(
|
|
id=company.official_address_id, db=db_session
|
|
).data
|
|
|
|
companies_list.append(
|
|
{
|
|
"uu_id": str(company.uu_id),
|
|
"public_name": company.public_name,
|
|
"company_type": company.company_type,
|
|
"company_address": company_address,
|
|
}
|
|
)
|
|
|
|
model_value = EmployeeTokenObject(
|
|
domain=domain,
|
|
user_type=UserType.employee.value,
|
|
user_uu_id=str(user.uu_id),
|
|
credentials=user.credentials(),
|
|
user_id=user.id,
|
|
person_id=user.person_id,
|
|
person_uu_id=str(user.person.uu_id),
|
|
request=dict(request.headers),
|
|
companies_uu_id_list=companies_uu_id_list,
|
|
companies_id_list=companies_id_list,
|
|
duty_uu_id_list=duty_uu_id_list,
|
|
duty_id_list=duty_id_list,
|
|
timezone=user.local_timezone or "GMT+0",
|
|
lang=user.lang or "tr",
|
|
).model_dump()
|
|
if cls.set_object_to_redis(user, model_value):
|
|
return {
|
|
"user_type": UserType.employee.name,
|
|
"companies_list": companies_list,
|
|
}
|
|
raise HTTPExceptionApi(
|
|
error_code="", lang="en", loc=get_line_number_for_error()
|
|
)
|
|
|
|
@classmethod
|
|
def remove_token_with_domain(cls, user: Users, domain: str) -> None:
|
|
"""Remove all tokens for a user with specific domain."""
|
|
redis_rows = cls._get_user_tokens(user)
|
|
for redis_row in redis_rows.all:
|
|
if redis_row.get("domain") == domain:
|
|
RedisActions.delete_key(redis_row.key)
|
|
|
|
@classmethod
|
|
def remove_all_token(cls, user: Users) -> None:
|
|
"""Remove all tokens for a user."""
|
|
redis_rows = cls._get_user_tokens(user)
|
|
RedisActions.delete([redis_row.key for redis_row in redis_rows.all])
|
|
|
|
@classmethod
|
|
def set_access_token_to_redis(
|
|
cls,
|
|
request: "Request",
|
|
user: Users,
|
|
domain: str,
|
|
remember: bool,
|
|
) -> Dict[str, Any]:
|
|
"""Set access token to redis and handle user session."""
|
|
cls.remove_token_with_domain(user=user, domain=domain)
|
|
Users.client_arrow = DateTimeLocal(is_client=True, timezone=user.local_timezone)
|
|
db_session = UsersTokens.new_session()
|
|
# Handle login based on user type
|
|
login_dict = (
|
|
cls.do_occupant_login(request=request, user=user, domain=domain)
|
|
if user.is_occupant
|
|
else (
|
|
cls.do_employee_login(request=request, user=user, domain=domain)
|
|
if user.is_employee
|
|
else {}
|
|
)
|
|
)
|
|
|
|
# Handle remember me functionality
|
|
if remember:
|
|
users_token = UsersTokens.find_or_create(
|
|
db=db_session,
|
|
user_id=user.id,
|
|
token_type="RememberMe",
|
|
token=cls._create_access_token(access=False),
|
|
domain=domain,
|
|
)
|
|
if users_token.meta_data.get("created"):
|
|
user.remember_me = True
|
|
else:
|
|
if UsersTokens.filter_all(
|
|
UsersTokens.user_id == user.id,
|
|
UsersTokens.token_type == "RememberMe",
|
|
UsersTokens.domain == domain,
|
|
db=db_session,
|
|
).data:
|
|
UsersTokens.filter_all(
|
|
UsersTokens.user_id == user.id,
|
|
UsersTokens.token_type == "RememberMe",
|
|
UsersTokens.domain == domain,
|
|
db=db_session,
|
|
).query.delete(synchronize_session=False)
|
|
user.remember_me = False
|
|
user.save(db=db_session)
|
|
return {
|
|
**login_dict,
|
|
"user": user.get_dict(),
|
|
}
|
|
|
|
@classmethod
|
|
def raise_error_if_request_has_no_token(cls, request: "Request") -> None:
|
|
"""Validate request has required token headers."""
|
|
if not hasattr(request, "headers"):
|
|
raise HTTPExceptionApi(
|
|
error_code="", lang="en", loc=get_line_number_for_error()
|
|
)
|
|
if not request.headers.get(Auth.ACCESS_TOKEN_TAG):
|
|
raise HTTPExceptionApi(
|
|
error_code="", lang="en", loc=get_line_number_for_error()
|
|
)
|
|
|
|
@classmethod
|
|
def access_token_is_valid(cls, request: "Request") -> bool:
|
|
"""Check if access token in request is valid."""
|
|
access_token = cls.get_access_token_from_request(request=request)
|
|
return RedisActions.get_json(
|
|
list_keys=AccessToken(accessToken=access_token).to_list()
|
|
).status
|
|
|
|
@classmethod
|
|
def get_access_token_from_request(cls, request: "Request") -> str:
|
|
"""Extract access token from request headers."""
|
|
cls.raise_error_if_request_has_no_token(request=request)
|
|
return request.headers.get(Auth.ACCESS_TOKEN_TAG)
|
|
|
|
@classmethod
|
|
def _process_redis_object(cls, redis_object: Dict[str, Any]) -> T:
|
|
"""Process Redis object and return appropriate token object."""
|
|
if not redis_object.get("selected_company"):
|
|
redis_object["selected_company"] = None
|
|
if not redis_object.get("selected_occupant"):
|
|
redis_object["selected_occupant"] = None
|
|
|
|
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 HTTPExceptionApi(
|
|
error_code="", lang="en", loc=get_line_number_for_error()
|
|
)
|
|
|
|
@classmethod
|
|
def get_object_via_access_key(cls, access_token: str) -> T:
|
|
"""Get token object using access key."""
|
|
access_token_obj = AccessToken(accessToken=access_token)
|
|
redis_response = RedisActions.get_json(list_keys=access_token_obj.to_list())
|
|
|
|
if redis_object := redis_response.first.data:
|
|
access_token_obj.userUUID = redis_object.get("user_uu_id")
|
|
return cls._process_redis_object(redis_object)
|
|
|
|
raise HTTPExceptionApi(
|
|
error_code="", lang="en", loc=get_line_number_for_error()
|
|
)
|
|
|
|
@classmethod
|
|
def get_object_via_user_uu_id(cls, user_id: str) -> T:
|
|
"""Get token object using user UUID."""
|
|
access_token = AccessToken(userUUID=user_id)
|
|
redis_response = RedisActions.get_json(list_keys=access_token.to_list())
|
|
|
|
if redis_object := redis_response.first.data:
|
|
access_token.userUUID = redis_object.get("user_uu_id")
|
|
return cls._process_redis_object(redis_object)
|
|
|
|
raise HTTPExceptionApi(
|
|
error_code="", lang="en", loc=get_line_number_for_error()
|
|
)
|