Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def aioredis_pool():
if sys.version_info >= (3, 5):
import aioredis
pool_coroutine = aioredis.create_redis_pool(
('localhost', 6379), minsize=2, maxsize=2)
return pool_coroutine, ring.aioredis
else:
pytest.skip()
async def test_connect_pool_aioredis_instance(self):
with patch('aioredlock.redis.Instance._create_redis_pool') as \
create_redis_pool:
redis_connection = await aioredis.create_redis_pool('redis://localhost')
instance = Instance(redis_connection)
await instance.connect()
assert not create_redis_pool.called
async def redisSecConnection():
try: # This Redis Index will be only for storage of security related items, like keys for examples
redisDB = await aioredis.create_redis_pool(("localhost", 6379), encoding="utf-8", db=10)
return redisDB
except OSError as ex:
print("Failed to connect to Redis, aborting action!")
raise ex
async def get_redis(cls) -> aioredis.Redis:
if cls.redis is not None:
return cls.redis
# TODO - we should let them override redis host/port in configuration
cls.redis = await aioredis.create_redis_pool((os.getenv('REDIS_HOST', 'localhost'), 6379), db=int(os.getenv('REDIS_DB', '1')), encoding='utf-8', minsize=1, maxsize=5)
return cls.redis
async def connect(self):
if self.closed:
self._redis = await aioredis.create_redis_pool((self.host, self.port), db=self.db)
async def main():
print("we are in [{}] mode".format(app_env))
app = web.Application()
app.add_routes(routes)
if app_env == 'development':
setup_swagger(
app,
swagger_url="/docs",
swagger_from_file="docs/main.yaml"
)
redis = await aioredis.create_redis_pool('redis://localhost', db=redis_db, minsize=5, maxsize=10, loop=app.loop)
app['redis'] = redis
return app
async def create_redis_pool(self, **kwargs):
return await aioredis.create_redis_pool(self.redis_dsn, loop=self.loop, **kwargs)
async def aio_redis_configure(_app, loop):
_c = dict(loop=loop)
if self.config:
config = self.config
else:
config = _app.config.get(self.config_name)
if not config:
raise ValueError("You must specify a redis_config or set the REDIS Sanic config variable")
if not isinstance(config, dict):
raise TypeError("Redis Config must be a dict")
for key in ['address', 'db', 'password', 'ssl', 'encoding', 'minsize',
'maxsize', 'timeout']:
if key in config:
_c.update({key: config.get(key)})
_redis = await create_redis_pool(**_c)
setattr(_app, self.config_name.lower(), _redis)
self.conn = _redis