evyos-center-server/production-evyos-systems-and-services-3
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

user selection compoleted tested

Browse Source
This commit is contained in:
Berkay
2025-06-15 19:05:48 +03:00
parent a48e560ece
commit 9fb517fa15
24 changed files with 574 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
ServicesApi/Extensions/OnMemory/redis_handlers.py
View File

@@ -33,7 +33,7 @@ class RedisHandlers:
@classmethod
def get_object_from_redis(cls, access_token: str) -> TokenDictType:
redis_response = RedisActions.get_json(list_keys=[cls.AUTH_TOKEN, access_token, "*", "*"])
redis_response = RedisActions.get_json(list_keys=[cls.AUTH_TOKEN, access_token, "*"])
if not redis_response.status:
raise ValueError("EYS_0001")
if redis_object := redis_response.first:
@@ -41,30 +41,27 @@ class RedisHandlers:
raise ValueError("EYS_0002")
@classmethod
def set_object_to_redis(cls, user: Users, token, header_info, add_uuid: str):
result_delete = RedisActions.delete(list_keys=[cls.AUTH_TOKEN, "*", str(user.uu_id), add_uuid])
def set_object_to_redis(cls, user: Users, token, header_info):
result_delete = RedisActions.delete(list_keys=[cls.AUTH_TOKEN, "*", str(user.uu_id)])
generated_access_token = PasswordModule.generate_access_token()
keys = [cls.AUTH_TOKEN, generated_access_token, str(user.uu_id)]
if add_uuid:
keys.append(add_uuid)
RedisActions.set_json(list_keys=keys, value={**token, **header_info}, expires={"hours": 1, "minutes": 30})
return generated_access_token
RedisActions.set_json(list_keys=keys, value={**token, **header_info}, expires={"hours": 1, "minutes": 30})
return generated_access_token
@classmethod
def update_token_at_redis(cls, token: str, add_payload: Union[CompanyToken, OccupantToken], add_uuid: str):
if already_token_data := RedisActions.get_json(list_keys=[cls.AUTH_TOKEN, token, '*', add_uuid]).first:
def update_token_at_redis(cls, token: str, add_payload: Union[dict], user_uuid: str):
list_keys = [cls.AUTH_TOKEN, token, "*"]
print("user_uuid", user_uuid, "add_payload", add_payload)
if not user_uuid:
raise ValueError("User UUID not found")
if already_token_data := RedisActions.get_json(list_keys=list_keys).first:
already_token = cls.process_redis_object(already_token_data)
if already_token.is_employee and isinstance(add_payload, CompanyToken):
if already_token.is_employee:
already_token.selected_company = add_payload
list_keys = [cls.AUTH_TOKEN, token, str(already_token.user_uu_id), str(add_uuid)]
print('is_employee: ', list_keys)
elif already_token.is_occupant and isinstance(add_payload, OccupantToken):
elif already_token.is_occupant:
already_token.selected_occupant = add_payload
list_keys = [cls.AUTH_TOKEN, token, str(already_token.user_uu_id), str(add_uuid)]
print('is_occupant: ', list_keys)
list_keys = [cls.AUTH_TOKEN, token, str(user_uuid)]
print("already_token", already_token)
result = RedisActions.set_json(list_keys=list_keys, value=already_token.model_dump(), expires={"hours": 1, "minutes": 30})
RedisActions.delete(list_keys=[cls.AUTH_TOKEN, token, str(already_token.user_uu_id)])
return result.first
raise ValueError("Something went wrong")