122 lines
3.0 KiB
Python
122 lines
3.0 KiB
Python
|
# 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()
|