mirror of
https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git
synced 2024-11-10 13:03:11 +00:00
refactor(backups): redis model storage utils
This commit is contained in:
parent
ea6a047478
commit
a0afe63b3d
30
selfprivacy_api/utils/redis_model_storage.py
Normal file
30
selfprivacy_api/utils/redis_model_storage.py
Normal file
|
@ -0,0 +1,30 @@
|
|||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
|
||||
def store_model_as_hash(redis, redis_key, model):
|
||||
for key, value in model.dict().items():
|
||||
if isinstance(value, datetime):
|
||||
value = value.isoformat()
|
||||
redis.hset(redis_key, key, str(value))
|
||||
|
||||
|
||||
def hash_as_model(redis, redis_key: str, model_class):
|
||||
token_dict = _model_dict_from_hash(redis, redis_key)
|
||||
if token_dict is not None:
|
||||
return model_class(**token_dict)
|
||||
return None
|
||||
|
||||
|
||||
def _prepare_model_dict(d: dict):
|
||||
for key in d.keys():
|
||||
if d[key] == "None":
|
||||
d[key] = None
|
||||
|
||||
|
||||
def _model_dict_from_hash(redis, redis_key: str) -> Optional[dict]:
|
||||
if redis.exists(redis_key):
|
||||
token_dict = redis.hgetall(redis_key)
|
||||
_prepare_model_dict(token_dict)
|
||||
return token_dict
|
||||
return None
|
36
tests/test_model_storage.py
Normal file
36
tests/test_model_storage.py
Normal file
|
@ -0,0 +1,36 @@
|
|||
import pytest
|
||||
|
||||
from pydantic import BaseModel
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from selfprivacy_api.utils.redis_model_storage import store_model_as_hash, hash_as_model
|
||||
from selfprivacy_api.utils.redis_pool import RedisPool
|
||||
|
||||
TEST_KEY = "model_storage"
|
||||
redis = RedisPool().get_connection()
|
||||
|
||||
@pytest.fixture()
|
||||
def clean_redis():
|
||||
redis.delete(TEST_KEY)
|
||||
|
||||
|
||||
class DummyModel(BaseModel):
|
||||
name: str
|
||||
date: Optional[datetime]
|
||||
|
||||
def test_store_retrieve():
|
||||
model = DummyModel(
|
||||
name= "test",
|
||||
date= datetime.now()
|
||||
)
|
||||
store_model_as_hash(redis, TEST_KEY, model)
|
||||
assert hash_as_model(redis, TEST_KEY, DummyModel) == model
|
||||
|
||||
def test_store_retrieve_none():
|
||||
model = DummyModel(
|
||||
name= "test",
|
||||
date= None
|
||||
)
|
||||
store_model_as_hash(redis, TEST_KEY, model)
|
||||
assert hash_as_model(redis, TEST_KEY, DummyModel) == model
|
Loading…
Reference in a new issue