Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_cache_types(self):
assert Cache.MEMORY == SimpleMemoryCache
assert Cache.REDIS == RedisCache
assert Cache.MEMCACHED == MemcachedCache
def test_parse_uri_path(self):
assert SimpleMemoryCache().parse_uri_path("/1/2/3") == {}
],
},
"alt": {"cache": "aiocache.SimpleMemoryCache"},
}
)
default = caches.create(**caches.get_alias_config("default"))
alt = caches.create(**caches.get_alias_config("alt"))
assert isinstance(default, RedisCache)
assert default.endpoint == "127.0.0.10"
assert default.port == 6378
assert isinstance(default.serializer, PickleSerializer)
assert len(default.plugins) == 2
assert isinstance(alt, SimpleMemoryCache)
def test_alias_takes_precedence(self, mock_cache):
with patch(
"aiocache.decorators.caches.get", MagicMock(return_value=mock_cache)
) as mock_get:
mc = multi_cached(
keys_from_attr="keys", alias="default", cache=SimpleMemoryCache, namespace="test"
)
mc(stub_dict)
mock_get.assert_called_with("default")
assert mc.cache is mock_cache
c = cached(
ttl=1,
key="key",
key_builder="fn",
cache=SimpleMemoryCache,
plugins=None,
alias=None,
noself=False,
namespace="test",
)
assert c.ttl == 1
assert c.key == "key"
assert c.key_builder == "fn"
assert c.cache is None
assert c._cache == SimpleMemoryCache
assert c._serializer is None
assert c._kwargs == {"namespace": "test"}
async def view_search_result(request):
cache = SimpleMemoryCache(serializer=JsonSerializer())
response_dict = await cache.get("response_dict")
if not response_dict:
response_dict = {}
return response_dict
logger = get_logger()
# The in-memory cache implementation stores data in a class member variable,
# so all instance of the in memory cache will reference that data structure.
# In order to separate transactions from devices, we need each cache to define
# its own namespace.
NS_TRANSACTION = 'synse.txn.'
NS_DEVICE = 'synse.dev.'
NS_ALIAS = 'synse.alias.'
transaction_cache = aiocache.SimpleMemoryCache(
namespace=NS_TRANSACTION,
)
device_cache = aiocache.SimpleMemoryCache(
namespace=NS_DEVICE,
)
alias_cache = aiocache.SimpleMemoryCache(
namespace=NS_ALIAS,
)
device_cache_lock = asyncio.Lock(loop=loop.synse_loop)
alias_cache_lock = asyncio.Lock(loop=loop.synse_loop)
async def get_transaction(transaction_id: str) -> dict:
"""Get the cached transaction information with the provided ID.
If the provided transaction ID does not correspond to a known transaction,
None is returned.
@cached(ttl=60*10, cache=SimpleMemoryCache, timeout=120)
async def nuls2_balance_getter(address, config=None):
global DECIMALS
if config is None:
from aleph.web import app
config = app['config']
server = get_server(config.nuls2.api_url.value)
contract_address = get_server(config.nuls2.contract_address.value)
chain_id = config.nuls2.chain_id.value
if DECIMALS is None:
response = await server.invokeView([chain_id, "decimals", "", []])
DECIMALS = int(response['result'])
response = await server.invokeView([chain_id, "balanceOf", "", [address]])
return int(response['result'])/(10**DECIMALS)
from structlog import get_logger
from synse_grpc import api
from synse_server import loop, plugin
logger = get_logger()
# The in-memory cache implementation stores data in a class member variable,
# so all instance of the in memory cache will reference that data structure.
# In order to separate transactions from devices, we need each cache to define
# its own namespace.
NS_TRANSACTION = 'synse.txn.'
NS_DEVICE = 'synse.dev.'
NS_ALIAS = 'synse.alias.'
transaction_cache = aiocache.SimpleMemoryCache(
namespace=NS_TRANSACTION,
)
device_cache = aiocache.SimpleMemoryCache(
namespace=NS_DEVICE,
)
alias_cache = aiocache.SimpleMemoryCache(
namespace=NS_ALIAS,
)
device_cache_lock = asyncio.Lock(loop=loop.synse_loop)
alias_cache_lock = asyncio.Lock(loop=loop.synse_loop)
async def get_transaction(transaction_id: str) -> dict:
# so all instance of the in memory cache will reference that data structure.
# In order to separate transactions from devices, we need each cache to define
# its own namespace.
NS_TRANSACTION = 'synse.txn.'
NS_DEVICE = 'synse.dev.'
NS_ALIAS = 'synse.alias.'
transaction_cache = aiocache.SimpleMemoryCache(
namespace=NS_TRANSACTION,
)
device_cache = aiocache.SimpleMemoryCache(
namespace=NS_DEVICE,
)
alias_cache = aiocache.SimpleMemoryCache(
namespace=NS_ALIAS,
)
device_cache_lock = asyncio.Lock(loop=loop.synse_loop)
alias_cache_lock = asyncio.Lock(loop=loop.synse_loop)
async def get_transaction(transaction_id: str) -> dict:
"""Get the cached transaction information with the provided ID.
If the provided transaction ID does not correspond to a known transaction,
None is returned.
Args:
transaction_id: The ID of the transaction.