cic-internal-integration/apps/cic-eth/cic_eth/server/cache.py

122 lines
3.0 KiB
Python
Raw Normal View History

2021-12-02 13:07:49 +01:00
# standard imports
import hashlib
import json
import logging
from typing import Optional, Union
# external imports
from cic_types.condiments import MetadataPointer
from redis import Redis
logg = logging.getLogger(__file__)
class Cache:
store: Redis = None
def get_token_data(token_symbol: str):
"""
:param token_symbol:
:type token_symbol:
:return:
:rtype:
"""
identifier = [token_symbol.encode('utf-8')]
key = cache_data_key(identifier, MetadataPointer.TOKEN_DATA)
logg.debug(f'Retrieving token data for: {token_symbol} at: {key}')
token_data_str = get_cached_data(key=key)
if(token_data_str is None):
logg.debug(f'No token data found for: {token_symbol}')
return None
else:
token_data = json.loads(token_data_str)
logg.debug(f'Retrieved token data: {token_data}')
return token_data
def set_token_data(token_symbol: str, data: dict):
"""
:param token_symbol:
:type token_symbol:
:return:
:rtype:
"""
identifier = [token_symbol.encode('utf-8')]
key = cache_data_key(identifier, MetadataPointer.TOKEN_DATA)
cache_data(key, json.dumps(data))
logg.debug(f'Cached token data for: {token_symbol} at: {key}')
def get_default_token() -> Optional[str]:
"""This function attempts to retrieve the default token's data from the redis cache.
:param chain_str: chain name and network id.
:type chain_str: str
:return:
:rtype:
"""
logg.debug(f'Retrieving default token from cache for chain:')
key = cache_data_key(identifier="ff".encode('utf-8'),
salt=MetadataPointer.TOKEN_DEFAULT)
token_data = json.loads(get_cached_data(key=key))
logg.debug(f'Retrieved default token data: {token_data}')
return token_data
def set_default_token(default_token: dict):
"""
:param default_token:
:type default_token:
:return:
:rtype:
"""
logg.debug(f'Setting default token in cache')
key = cache_data_key(identifier="ff".encode('utf-8'),
salt=MetadataPointer.TOKEN_DEFAULT)
cache_data(key, json.dumps(default_token))
def cache_data(key: str, data: str):
"""
:param key:
:type key:
:param data:
:type data:
:return:
:rtype:
"""
cache = Cache.store
cache.set(name=key, value=data)
cache.persist(name=key)
logg.debug(f'caching: {data} with key: {key}.')
def get_cached_data(key: str):
"""
:param key:
:type key:
:return:
:rtype:
"""
cache = Cache.store
return cache.get(name=key)
def cache_data_key(identifier: Union[list, bytes], salt: MetadataPointer):
"""
:param identifier:
:type identifier:
:param salt:
:type salt:
:return:
:rtype:
"""
hash_object = hashlib.new("sha256")
if isinstance(identifier, list):
for identity in identifier:
hash_object.update(identity)
else:
hash_object.update(identifier)
hash_object.update(salt.value.encode(encoding="utf-8"))
return hash_object.digest().hex()