Add additional, typesafe getters
This commit is contained in:
parent
7751768b2e
commit
7ff30c6df8
@ -44,27 +44,6 @@ class KeyValueStore():
|
||||
Supports the types str, datetime, float and int.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def get_value(key: str) -> Optional[ValueTypes]:
|
||||
"""
|
||||
Get the value for the given key.
|
||||
:param key: Key to get the value for
|
||||
"""
|
||||
kv = _KeyValueStoreModel.session.query(_KeyValueStoreModel).filter(
|
||||
_KeyValueStoreModel.key == key).first()
|
||||
if kv is None:
|
||||
return None
|
||||
if kv.value_type == ValueTypesEnum.STRING:
|
||||
return kv.string_value
|
||||
if kv.value_type == ValueTypesEnum.DATETIME and kv.datetime_value is not None:
|
||||
return kv.datetime_value.replace(tzinfo=timezone.utc)
|
||||
if kv.value_type == ValueTypesEnum.FLOAT:
|
||||
return kv.float_value
|
||||
if kv.value_type == ValueTypesEnum.INT:
|
||||
return kv.int_value
|
||||
# This should never happen unless someone messed with the database manually
|
||||
raise ValueError(f'Unknown value type {kv.value_type}') # pragma: no cover
|
||||
|
||||
@staticmethod
|
||||
def store_value(key: str, value: ValueTypes) -> None:
|
||||
"""
|
||||
@ -105,6 +84,79 @@ class KeyValueStore():
|
||||
_KeyValueStoreModel.session.delete(kv)
|
||||
_KeyValueStoreModel.session.commit()
|
||||
|
||||
@staticmethod
|
||||
def get_value(key: str) -> Optional[ValueTypes]:
|
||||
"""
|
||||
Get the value for the given key.
|
||||
:param key: Key to get the value for
|
||||
"""
|
||||
kv = _KeyValueStoreModel.session.query(_KeyValueStoreModel).filter(
|
||||
_KeyValueStoreModel.key == key).first()
|
||||
if kv is None:
|
||||
return None
|
||||
if kv.value_type == ValueTypesEnum.STRING:
|
||||
return kv.string_value
|
||||
if kv.value_type == ValueTypesEnum.DATETIME and kv.datetime_value is not None:
|
||||
return kv.datetime_value.replace(tzinfo=timezone.utc)
|
||||
if kv.value_type == ValueTypesEnum.FLOAT:
|
||||
return kv.float_value
|
||||
if kv.value_type == ValueTypesEnum.INT:
|
||||
return kv.int_value
|
||||
# This should never happen unless someone messed with the database manually
|
||||
raise ValueError(f'Unknown value type {kv.value_type}') # pragma: no cover
|
||||
|
||||
@staticmethod
|
||||
def get_string_value(key: str) -> Optional[str]:
|
||||
"""
|
||||
Get the value for the given key.
|
||||
:param key: Key to get the value for
|
||||
"""
|
||||
kv = _KeyValueStoreModel.session.query(_KeyValueStoreModel).filter(
|
||||
_KeyValueStoreModel.key == key,
|
||||
_KeyValueStoreModel.value_type == ValueTypesEnum.STRING).first()
|
||||
if kv is None:
|
||||
return None
|
||||
return kv.string_value
|
||||
|
||||
@staticmethod
|
||||
def get_datetime_value(key: str) -> Optional[datetime]:
|
||||
"""
|
||||
Get the value for the given key.
|
||||
:param key: Key to get the value for
|
||||
"""
|
||||
kv = _KeyValueStoreModel.session.query(_KeyValueStoreModel).filter(
|
||||
_KeyValueStoreModel.key == key,
|
||||
_KeyValueStoreModel.value_type == ValueTypesEnum.DATETIME).first()
|
||||
if kv is None or kv.datetime_value is None:
|
||||
return None
|
||||
return kv.datetime_value.replace(tzinfo=timezone.utc)
|
||||
|
||||
@staticmethod
|
||||
def get_float_value(key: str) -> Optional[float]:
|
||||
"""
|
||||
Get the value for the given key.
|
||||
:param key: Key to get the value for
|
||||
"""
|
||||
kv = _KeyValueStoreModel.session.query(_KeyValueStoreModel).filter(
|
||||
_KeyValueStoreModel.key == key,
|
||||
_KeyValueStoreModel.value_type == ValueTypesEnum.FLOAT).first()
|
||||
if kv is None:
|
||||
return None
|
||||
return kv.float_value
|
||||
|
||||
@staticmethod
|
||||
def get_int_value(key: str) -> Optional[int]:
|
||||
"""
|
||||
Get the value for the given key.
|
||||
:param key: Key to get the value for
|
||||
"""
|
||||
kv = _KeyValueStoreModel.session.query(_KeyValueStoreModel).filter(
|
||||
_KeyValueStoreModel.key == key,
|
||||
_KeyValueStoreModel.value_type == ValueTypesEnum.INT).first()
|
||||
if kv is None:
|
||||
return None
|
||||
return kv.int_value
|
||||
|
||||
|
||||
def set_startup_time():
|
||||
"""
|
||||
|
@ -17,9 +17,18 @@ def test_key_value_store(time_machine):
|
||||
KeyValueStore.store_value("test_int", 15)
|
||||
|
||||
assert KeyValueStore.get_value("test") == "testStringValue"
|
||||
assert KeyValueStore.get_value("test") == "testStringValue"
|
||||
assert KeyValueStore.get_string_value("test") == "testStringValue"
|
||||
assert KeyValueStore.get_value("test_dt") == datetime.now(timezone.utc)
|
||||
assert KeyValueStore.get_datetime_value("test_dt") == datetime.now(timezone.utc)
|
||||
assert KeyValueStore.get_string_value("test_dt") is None
|
||||
assert KeyValueStore.get_float_value("test_dt") is None
|
||||
assert KeyValueStore.get_int_value("test_dt") is None
|
||||
assert KeyValueStore.get_value("test_float") == 22.51
|
||||
assert KeyValueStore.get_float_value("test_float") == 22.51
|
||||
assert KeyValueStore.get_value("test_int") == 15
|
||||
assert KeyValueStore.get_int_value("test_int") == 15
|
||||
assert KeyValueStore.get_datetime_value("test_int") is None
|
||||
|
||||
time_machine.move_to(start + timedelta(days=20, hours=5), tick=False)
|
||||
assert KeyValueStore.get_value("test_dt") != datetime.now(timezone.utc)
|
||||
|
Loading…
Reference in New Issue
Block a user