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

feat(tokens-repo): redis get tokens

Browse source
This commit is contained in:
Houkime 2022-12-14 14:48:43 +00:00
parent f2fa47466b
commit 256c16fa9f
1 changed files with 20 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -28,7 +28,9 @@ class RedisTokensRepository(AbstractTokensRepository):
def get_tokens(self) -> list[Token]: def get_tokens(self) -> list[Token]:
"""Get the tokens""" """Get the tokens"""
raise NotImplementedError r = self.connection
token_keys = r.keys(TOKENS_PREFIX + "*")
return [self._token_from_hash(key) for key in token_keys]
def delete_token(self, input_token: Token) -> None: def delete_token(self, input_token: Token) -> None:
"""Delete the token""" """Delete the token"""
@ -65,3 +67,20 @@ class RedisTokensRepository(AbstractTokensRepository):
def _get_stored_new_device_key(self) -> Optional[NewDeviceKey]: def _get_stored_new_device_key(self) -> Optional[NewDeviceKey]:
"""Retrieves new device key that is already stored.""" """Retrieves new device key that is already stored."""
raise NotImplementedError raise NotImplementedError
def _token_from_hash(self, redis_key: str) -> Token:
r = self.connection
if r.exists(redis_key):
token_dict = r.hgetall(redis_key)
for date in [
"created_at",
]:
if token_dict[date] != "None":
token_dict[date] = datetime.datetime.fromisoformat(token_dict[date])
for key in token_dict.keys():
if token_dict[key] == "None":
token_dict[key] = None
return Token(**token_dict)
return None