SelfPrivacy/selfprivacy-rest-api
1
0
Fork 0
mirror of https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git synced 2025-01-11 18:39:30 +00:00
Code Issues Projects Releases Wiki Activity

feat(tokens-repo): get recovery key

Browse source
This commit is contained in:
Houkime 2022-12-14 18:01:34 +00:00
parent 257096084f
commit 4579fec569
1 changed files with 12 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
selfprivacy_api/repositories/tokens/redis_tokens_repository.py
View file

@ -15,6 +15,7 @@ from selfprivacy_api.repositories.tokens.exceptions import TokenNotFound
TOKENS_PREFIX = "token_repo:tokens:" TOKENS_PREFIX = "token_repo:tokens:"
NEW_DEVICE_KEY_REDIS_KEY = "token_repo:new_device_key" NEW_DEVICE_KEY_REDIS_KEY = "token_repo:new_device_key"
RECOVERY_KEY_REDIS_KEY = "token_repo:recovery_key"
class RedisTokensRepository(AbstractTokensRepository): class RedisTokensRepository(AbstractTokensRepository):
@ -49,7 +50,10 @@ class RedisTokensRepository(AbstractTokensRepository):
def get_recovery_key(self) -> Optional[RecoveryKey]: def get_recovery_key(self) -> Optional[RecoveryKey]:
"""Get the recovery key""" """Get the recovery key"""
raise NotImplementedError r = self.connection
if r.exists(RECOVERY_KEY_REDIS_KEY):
return self._recovery_key_from_hash(RECOVERY_KEY_REDIS_KEY)
return None
def create_recovery_key( def create_recovery_key(
self, self,
@ -90,6 +94,7 @@ class RedisTokensRepository(AbstractTokensRepository):
def _is_date_key(key: str): def _is_date_key(key: str):
return key in [ return key in [
"created_at", "created_at",
"expires_at",
] ]
@staticmethod @staticmethod
@ -116,6 +121,12 @@ class RedisTokensRepository(AbstractTokensRepository):
return Token(**token_dict) return Token(**token_dict)
return None return None
def _recovery_key_from_hash(self, redis_key: str) -> Optional[RecoveryKey]:
token_dict = self._model_dict_from_hash(redis_key)
if token_dict is not None:
return RecoveryKey(**token_dict)
return None
def _store_model_as_hash(self, redis_key, model): def _store_model_as_hash(self, redis_key, model):
r = self.connection r = self.connection
for key, value in model.dict().items(): for key, value in model.dict().items():