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_settings(self):
settings.set_cache(
RedisCache, endpoint="127.0.0.1", port=6379, timeout=10, db=1)
cache = RedisCache(db=0)
assert cache.endpoint == "127.0.0.1"
assert cache.port == 6379
assert cache.timeout == 10
assert cache.db == 0
def test_get_cache_value_with_fallbacks_with_settings():
settings.set_cache(RedisCache, fake_key="random")
assert get_cache_value_with_fallbacks(None, "fake_key", "DEFAULT", cls=RedisCache) == "random"
def test_get_alias(self):
assert settings.get_alias("default") == {
'cache': "aiocache.SimpleMemoryCache",
'serializer': {
'class': "aiocache.serializers.StringSerializer"
}
def test_settings_singleton(self):
assert settings() is settings()
@cached(ttl=259200, key_from_attr='novels_name', serializer=PickleSerializer(), namespace="novels_name")
async def start(novels_name):
"""
Start spider
:return:
"""
return await SoNovels.start(novels_name)
if __name__ == '__main__':
# Start
import aiocache
REDIS_DICT = {}
aiocache.settings.set_defaults(
class_="aiocache.RedisCache",
endpoint=REDIS_DICT.get('REDIS_ENDPOINT', 'localhost'),
port=REDIS_DICT.get('REDIS_PORT', 6379),
db=REDIS_DICT.get('CACHE_DB', 0),
password=REDIS_DICT.get('REDIS_PASSWORD', None),
)
res = asyncio.get_event_loop().run_until_complete(start('雪中悍刀行 小说 最新章节'))
print(res)
def init_cache(app, loop):
LOGGER.info("Starting aiocache")
app.config.from_object(CONFIG)
REDIS_DICT = CONFIG.REDIS_DICT
aiocache.settings.set_defaults(
class_="aiocache.RedisCache",
endpoint=REDIS_DICT.get('REDIS_ENDPOINT', 'localhost'),
port=REDIS_DICT.get('REDIS_PORT', 6379),
db=REDIS_DICT.get('CACHE_DB', 0),
password=REDIS_DICT.get('REDIS_PASSWORD', None),
loop=loop,
)
LOGGER.info("Starting redis pool")
redis_session = RedisSession()
# redis instance for app
app.get_redis_pool = redis_session.get_redis_pool
# pass the getter method for the connection pool into the session
app.session_interface = RedisSessionInterface(
app.get_redis_pool, cookie_name="owl_sid", expiry=30 * 24 * 60 * 60)
@cached(ttl=259200, key_from_attr='novels_name', serializer=PickleSerializer(), namespace="novels_name")
async def start(novels_name):
"""
Start spider
:return:
"""
return await BingNovels.start(novels_name)
if __name__ == '__main__':
# Start
import aiocache
REDIS_DICT = {}
aiocache.settings.set_defaults(
class_="aiocache.RedisCache",
endpoint=REDIS_DICT.get('REDIS_ENDPOINT', 'localhost'),
port=REDIS_DICT.get('REDIS_PORT', 6379),
db=REDIS_DICT.get('CACHE_DB', 0),
password=REDIS_DICT.get('REDIS_PASSWORD', None),
)
res = asyncio.get_event_loop().run_until_complete(start('雪中悍刀行 小说 阅读 最新章节'))
print(res)
def get_cache(cache=None, serializer=None, plugins=None, **kwargs):
cache = cache or settings.get_cache_class()
serializer = serializer
plugins = plugins
instance = cache(
serializer=serializer,
plugins=plugins,
**{**settings.get_cache_args(), **kwargs})
return instance
def get_cache(cache=None, serializer=None, plugins=None, **kwargs):
cache = cache or settings.get_cache_class()
serializer = serializer
plugins = plugins
instance = cache(
serializer=serializer,
plugins=plugins,
**{**settings.get_cache_args(), **kwargs})
return instance