import json from api_services.redis.conn import redis_cli class RedisResponse: def __init__(self, status: bool, message: str, data: dict = None, error: str = None): self.status = status self.message = message self.data = data self.error = error def as_dict(self): return { "status": self.status, "message": self.message, "data": self.data, "error": self.error, } class RedisActions: @classmethod def set_json(cls, name, value): try: search_name = str(name) if isinstance(name, str) else name.decode() redis_cli.set(name=search_name, value=json.dumps(value)) return RedisResponse( status=True, message="Value is set successfully.", data=value, ) except Exception as e: return RedisResponse( status=False, message="Value is not set successfully.", error=str(e), ) @classmethod def get_json(cls, name): try: search_name = str(name) if isinstance(name, str) else name.decode() json_get = redis_cli.get(search_name) if not json_get: return RedisResponse( status=False, message="Value is not get successfully.", error="Value is not found in the redis.", ) return RedisResponse( status=True, message="Value is get successfully.", data=json.loads(json_get), ) except Exception as e: return RedisResponse( status=False, message="Value is not get successfully.", error=str(e), ) @classmethod def set_replace_all(cls, value, value_regex): try: already_tokens = redis_cli.scan_iter(match=str(value_regex)) for already_token in already_tokens: redis_cli.set(name=already_token, value=json.dumps(value)) return RedisResponse( status=True, message="Value is set successfully.", data=value, ) except Exception as e: return RedisResponse( status=False, message="Value is not set successfully.", error=str(e), ) @classmethod def delete(cls, name): try: search_name = str(name) if isinstance(name, str) else name.decode() json_delete = redis_cli.delete(search_name) if not json_delete: return RedisResponse( status=False, message="Value is not deleted successfully.", error="Value is not found in the redis.", ) return RedisResponse( status=True, message="Value is deleted successfully.", ) except Exception as e: return RedisResponse( status=False, message="Value is not deleted successfully.", error=str(e), ) @classmethod def delete_all(cls, value_regex): try: already_tokens = redis_cli.scan_iter(match=str(value_regex)) for already_token in already_tokens: redis_cli.delete(already_token) return RedisResponse( status=True, message="Value is deleted successfully.", ) except Exception as e: return RedisResponse( status=False, message="Value is not deleted successfully.", error=str(e), ) @classmethod def update(cls, name, data): try: json_update = cls.get_json(name=name) if json_update.status: value_dict = json_update.data for key, value in data.items(): value_dict[key] = value redis_cli.set(name=name, value=json.dumps(value_dict)) return RedisResponse( status=True, message="Value is updated successfully.", data=value_dict, ) except Exception as e: return RedisResponse( status=False, message="Value is not updated successfully.", error=str(e), ) @classmethod def update_all(cls, value_regex, data): try: already_tokens = redis_cli.scan_iter(match=str(value_regex)) for already_token in already_tokens: cls.update(name=already_token, data=data) return RedisResponse( status=True, message="Values are updated successfully.", data=data, ) except Exception as e: return RedisResponse( status=False, message="Value is not updated successfully.", error=str(e), ) # def get_object_via_access_key( # request, # ): # # if not hasattr(request, "headers"): # raise redis_imports.exceptions( # status_code=401, # detail=dict( # message="Headers are not found in request. Invalid request object." # ), # ) # if not request.headers.get(redis_imports.ACCESS_TOKEN_TAG): # raise redis_imports.exceptions( # status_code=401, # detail=dict(message="Unauthorized user, please login..."), # ) # already_tokens = redis_cli.scan_iter( # match=str(request.headers.get(redis_imports.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 redis_imports.EmployeeTokenObject(**redis_object) # elif redis_object.get("user_type") == 2: # if not redis_object.get("selected_occupant", None): # redis_object["selected_occupant"] = None # return redis_imports.OccupantTokenObject(**redis_object) # raise redis_imports.exceptions( # 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 redis_imports.exceptions( # status_code=500, # detail={ # "message": "Redis Service raised an exception.", # "error": str(e), # }, # ) # # raise redis_imports.exceptions( # status_code=redis_imports.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 # # # def save_object_to_redis( # access_token, model_object, redis_imports: RedisImports # ) -> bool: # try: # if redis_cli.set( # name=str(access_token) + ":" + str(model_object.user_uu_id), # value=model_object.model_dump_json(), # ): # return access_token # except Exception as e: # print("Save Object to Redis Error: ", e) # raise redis_imports.exceptions( # status_code=redis_imports.status.HTTP_503_SERVICE_UNAVAILABLE, # detail=dict( # message="Headers are not found in request. Invalid request object. Redis Error: Token is not saved." # ), # )