Python aioredis 模块，create_redis() 实例源码

def get_app():
    redis = await aioredis.create_redis(('localhost', 6379,), db=1)

    app = web.Application()
    app['redis'] = redis

    aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader('templates/'),
                         context_processors=(static_processor,))

    app.router.add_route('GET', '/', handlers.index)
    app.router.add_route('GET', '/login', handlers.login_task)
    app.router.add_route('POST', '/login', handlers.login)
    app.router.add_static('/static', 'static')

    async def close_redis(app):
        app['redis'].close()

    app.on_shutdown.append(close_redis)

    return app

def run_bot():
    if BotConfig.check():
        loop = asyncio.get_event_loop()

        logger.info('creating redis connection')
        redis_conn = loop.run_until_complete(aioredis.create_redis(('localhost', 6379), encoding="utf-8"))

        logger.info('adding signal handlers')
        add_signal_handlers(loop, redis_conn)

        logger.info('creating tasks: bot and background coros')
        create_tasks(loop, redis_conn)

        try:
            logger.info('starting event loop ')
            loop.run_forever()
        finally:
            loop.close()

def test_send_message(self):
        async def func():
            SUB = await aioredis.create_redis((os.environ.get(
                "REDIS_HOST",
                "127.0.0.1"
            ), 6379))
            res = await SUB.psubscribe("enibar-*")
            subscriber = res[0]

            api.redis.send_message('enibar-test', 'test')

            await subscriber.wait_message()
            reply = await subscriber.get_json()
            self.assertEqual(reply, (b'enibar-test', 'test'))

        task = asyncio.ensure_future(func())
        self.loop.run_until_complete(task)

def store_interface(config, interface, key=None, exist='SET_IF_NOT_EXIST'):
    redis = yield from aioredis.create_redis((config.redis_host, config.redis_port))
    pipe = redis.pipeline()
    interface_id = interface['network_interface_id']

    # Only store picked interface data if using default key (not fixed key from instance)
    if not key:
        key = KEY_ENI + interface_id
        pipe.set(key=KEY_ENI + interface_id, value=pickle.dumps(interface), expire=int(config.redis_ttl))

    # Store intermediate key lookups so that we can find metadata given only an IP address
    if 'association' in interface:
        pipe.set(key=KEY_IP + interface['association']['public_ip'], value=key, expire=int(config.redis_ttl), exist=exist)

    for address in interface.get('private_ip_addresses', []):
        pipe.set(key=KEY_IP + address['private_ip_address'], value=key, expire=int(config.redis_ttl), exist=exist)

    yield from pipe.execute()
    redis.close()
    yield from redis.wait_closed()

def __init__(self, io_loop: asyncio.AbstractEventLoop = None):
        super().__init__()
        self.io_loop = io_loop or asyncio.get_event_loop()
        self.sub_client = self.io_loop.run_until_complete(
                aioredis.create_redis((config.get('REDIS', 'host', fallback='localhost'),
                                       config.getint('REDIS', 'port', fallback=6379)),
                                      db=config.getint('REDIS', 'db', fallback=1)))
        self.redis_client = redis.StrictRedis(
            host=config.get('REDIS', 'host', fallback='localhost'),
            db=config.getint('REDIS', 'db', fallback=1), decode_responses=True)
        self.initialized = False
        self.sub_tasks = list()
        self.sub_channels = list()
        self.channel_router = dict()
        self.crontab_router = defaultdict(dict)
        self.datetime = None
        self.time = None
        self.loop_time = None

def setup(self):
        self.redis = await aioredis.create_redis((config.REDIS_HOST, config.REDIS_PORT,), encoding='utf8')
        self.__processed_status = await self.get_processed_ids()
        self.client = aiohttp.ClientSession()

        extension_manager = extension.ExtensionManager(namespace='ofensivaria.bot.commands',
                                                       invoke_on_load=True,
                                                       invoke_args=(self, self.redis, self.client))

        commands = extension_manager.map(self.__extension_manager_callback)
        self.commands = [obj for name, obj in sorted(commands)]

        prepare_tasks = [c.prepare() for c in self.commands]
        await asyncio.gather(*prepare_tasks)

        self.__setup = True

def connect(self, endpoint: str, auth: Optional[str] = None) -> None:  # type: ignore
        endpoint = validate_endpoint(endpoint)
        auth = validate_auth(auth)
        self._auth = auth
        # print('*** redis connecting to ', endpoint, flush=True)
        if self._connected:
            raise ConnectionError('Already connected.')
        if not endpoint.startswith('redis://'):
            raise ValueError('Expected endpoint to begin with "redis://".'
                             'Got: {!r}'.format(endpoint))
        host, port = endpoint.replace('redis://', '').split(':')  # todo: handle exception
        self._subscriber = await aioredis.create_redis((host, port))
        self._publisher = await aioredis.create_redis((host, port))
        if auth:
            await self._subscriber.auth(auth)
            await self._publisher.auth(auth)
        else:
            print('*** WARNING: Redis connection has no password.')
        self._connected = True

def redis_conn(loop):
    """
    yield fixture which creates a redis connection, and flushes redis before the test.

    Note: redis is not flushed after the test both for performance and to allow later debugging.
    """
    async def _get_conn():
        conn = await create_redis(('localhost', 6379), loop=loop)
        await conn.flushall()
        return conn
    conn = loop.run_until_complete(_get_conn())
    conn.loop = loop
    yield conn

    conn.close()
    try:
        loop.run_until_complete(conn.wait_closed())
    except RuntimeError:
        pass

def prepare(self):
        """
        Called by the backend to prepare SocketShark (i.e. initialize Redis
        connection and the receiver class)
        """
        redis_receiver = Receiver(loop=asyncio.get_event_loop())
        redis_settings = self.config['REDIS']
        try:
            self.redis = await aioredis.create_redis((
                redis_settings['host'], redis_settings['port']))
        except (OSError, aioredis.RedisError):
            self.log.exception('could not connect to redis')
            raise

        # Some features (e.g. pinging) don't work on old Redis versions.
        info = await self.redis.info('server')
        version_info = info['server']['redis_version'].split('.')
        major, minor = int(version_info[0]), int(version_info[1])
        if not (major > 3 or major == 3 and minor >= 2):
            msg = 'Redis version must be at least 3.2'
            self.log.exception(msg, version_info=version_info)
            raise Exception(msg)

        self._redis_connection_handler_task = asyncio.ensure_future(
                self._redis_connection_handler())

        self.service_receiver = ServiceReceiver(self, redis_receiver)

def __init__(self,
                 dsn: str = "redis://127.0.0.1:6379/0",
                 prefix: str = "aiotasks",
                 loop=None):
        super().__init__(loop=loop, prefix=prefix)

        _, password, host, port, db = parse_dsn(dsn,
                                                default_port=6379,
                                                default_db=0)
        db = int(db)

        # if not port:
        #     port = 6379
        #
        # port = int(port)
        # try:
        #     db = int(db)
        #
        #     if not db:
        #         db = 0
        # except ValueError:
        #     db = 0

        self._redis_pub = self._loop_subscribers.run_until_complete(
            aioredis.create_redis(address=(host, port),
                                  db=db,
                                  password=password,
                                  loop=self._loop_subscribers))

        self._redis_sub = self._loop_subscribers.run_until_complete(
            aioredis.create_redis(address=(host, port),
                                  db=db,
                                  password=password,
                                  loop=self._loop_subscribers))

def __init__(self,
                 dsn: str = "redis://127.0.0.1:6379/0",
                 prefix: str = "aiotasks",
                 concurrency: int = 5,
                 loop=None):
        super().__init__(loop=loop, prefix=prefix, concurrency=concurrency)

        _, password, host, port, db = parse_dsn(dsn,
                                                default_port=6379,
                                                default_db=0)

        db = int(db)

        # if not port:
        #     port = 6379
        #
        # port = int(port)
        # try:
        #     db = int(db)
        #     if not db:
        #         db = 0
        # except ValueError:
        #     db = 0

        self._redis_consumer = self._loop_delay. \
            run_until_complete(aioredis.create_redis(address=(host, port),
                                                     db=db,
                                                     password=password,
                                                     loop=self._loop_delay))

        self._redis_poller = self._loop_delay. \
            run_until_complete(aioredis.create_redis(address=(host, port),
                                                     db=db,
                                                     password=password,
                                                     loop=self._loop_delay))

def on_connect(self):
        """Initialize Redis connection when bot loads"""
        self.redis = await aioredis.create_redis(('localhost', 6379))

def get_async_redis(self, loop=None):
        """Creates an asynchronous Redis connection.

        Parameters
        ----------
        loop = Optional[asyncio.AbstractEventLoop]
            The loop used for the asynchronous Redis connection.
        """

        if self.loop is not None and loop is None:
            loop = self.loop
        return await aioredis.create_redis(
            'redis://{}:{}'.format(cache_config['HOST'], cache_config['PORT']),
            db=cache_config['DB'], password=cache_config['PASSWORD'], loop=loop)

def redis(variables, loop):
    coro = create_redis(
        (variables['redis']['host'], variables['redis']['port']),
        db=variables['redis']['db'],
        loop=loop
    )
    return loop.run_until_complete(coro)

def get_async_redis():
    """ initialize an asyncronous redis connection
    """
    global ASYNCREDIS
    if ASYNCREDIS is None or ASYNCREDIS.closed:  # pragma: no branch
        address = REDIS_PUBSUB["address"]
        db = REDIS_PUBSUB["db"]
        password = REDIS_PUBSUB["password"]
        ASYNCREDIS = yield from aioredis.create_redis(address, db=db, password=password)
    return ASYNCREDIS

def redis_async(request, event_loop):
    conn = event_loop.run_until_complete(aioredis.create_redis(('localhost', 6379), encoding="utf-8", db=10))

    def redis_async_cleanup():
        conn.close()
    request.addfinalizer(redis_async_cleanup)
    return conn

def redis(variables, loop):
    coro = create_redis(
        (variables['redis']['host'], variables['redis']['port']),
        db=variables['redis']['db'],
        loop=loop
    )
    return loop.run_until_complete(coro)

def _publish(self, data):
        if self.pub is None:
            self.pub = await aioredis.create_redis((self.host, self.port),
                                                   db=self.db)
        return await self.pub.publish(self.channel, pickle.dumps(data))

def _listen(self):
        if self.sub is None:
            self.sub = await aioredis.create_redis((self.host, self.port),
                                                   db=self.db)
            self.ch = (await self.sub.subscribe(self.channel))[0]
        while True:
            return await self.ch.get()

def install_redis_handle(app):
    global SUB
    while True:
        SUB = await aioredis.create_redis((settings.REDIS_HOST, 6379), password=settings.REDIS_PASSWORD)
        res = await SUB.psubscribe("enibar-*")
        subscriber = res[0]

        while await subscriber.wait_message():
            reply = await subscriber.get_json()
            await app.redis_handle(reply[0].decode(), reply[1])
        await asyncio.sleep(1)

def connect_databases(self):
        # connect to postgres
        self.pgpool = await asyncpg.create_pool(**self.cfg['db']['postgres'])

        # connect to redis
        self.redis = await aioredis.create_redis(
            (self.cfg['db']['redis'], 6379), loop=self.loop
        )

def init_connections(sanic, loop):
    """Bind the database and Redis client to Sanic's event loop."""

    global redis
    redis_server = environ.get('REDIS_SERVER', None)
    if redis_server:
         redis = await create_redis((redis_server, 6379), encoding='utf-8', loop=loop)
    seed()

def store_instance(config, instance):
    redis = yield from aioredis.create_redis((config.redis_host, config.redis_port))
    pipe = redis.pipeline()
    instance_id = instance['instance_id']

    # Store pickled instance data keyed off instance ID
    pipe.set(key=KEY_I + instance_id, value=pickle.dumps(instance), expire=int(config.redis_ttl))

    # Store intermediate key lookups so that we can find an instance given only its IP address
    for interface in instance.get('network_interfaces', []):
        yield from store_interface(config, interface, KEY_I + instance_id, None)

    yield from pipe.execute()
    redis.close()
    yield from redis.wait_closed()

def create():
        return Database(await aioredis.create_redis(("redis", 6379)))

def test_aioredis(i, loop):
    start = time.time()
    redis = await aioredis.create_redis((HOST, 6379), loop=loop)
    val = None
    for i in range(i):
        val = await redis.keys('*')
    print(time.time() - start)
    redis.close()
    await redis.wait_closed()
    return val

def query(self, query_type: str, **kwargs):
        sub_client = None
        channel_name1, channel_name2 = None, None
        try:
            sub_client = await aioredis.create_redis(
                (config.get('REDIS', 'host', fallback='localhost'),
                 config.getint('REDIS', 'port', fallback=6379)),
                db=config.getint('REDIS', 'db', fallback=1))
            request_id = self.next_id()
            kwargs['RequestID'] = request_id
            channel_name1 = self.__trade_response_format.format('OnRspQry' + query_type, request_id)
            channel_name2 = self.__trade_response_format.format('OnRspError', request_id)
            ch1, ch2 = await sub_client.psubscribe(channel_name1, channel_name2)
            cb = self.io_loop.create_future()
            tasks = [
                asyncio.ensure_future(self.query_reader(ch1, cb), loop=self.io_loop),
                asyncio.ensure_future(self.query_reader(ch2, cb), loop=self.io_loop),
            ]
            self.redis_client.publish(self.__request_format.format('ReqQry' + query_type), json.dumps(kwargs))
            rst = await asyncio.wait_for(cb, HANDLER_TIME_OUT, loop=self.io_loop)
            await sub_client.punsubscribe(channel_name1, channel_name2)
            sub_client.close()
            await asyncio.wait(tasks, loop=self.io_loop)
            return rst
        except Exception as e:
            logger.error('%s failed: %s', query_type, repr(e), exc_info=True)
            if sub_client and sub_client.in_pubsub and channel_name1:
                await sub_client.unsubscribe(channel_name1, channel_name2)
                sub_client.close()
            return None

def SubscribeMarketData(self, inst_ids: list):
        sub_client = None
        channel_name1, channel_name2 = None, None
        try:
            sub_client = await aioredis.create_redis(
                (config.get('REDIS', 'host', fallback='localhost'),
                 config.getint('REDIS', 'port', fallback=6379)),
                db=config.getint('REDIS', 'db', fallback=1))
            channel_name1 = self.__market_response_format.format('OnRspSubMarketData', 0)
            channel_name2 = self.__market_response_format.format('OnRspError', 0)
            ch1, ch2 = await sub_client.psubscribe(channel_name1, channel_name2)
            cb = self.io_loop.create_future()
            tasks = [
                asyncio.ensure_future(self.query_reader(ch1, cb), loop=self.io_loop),
                asyncio.ensure_future(self.query_reader(ch2, cb), loop=self.io_loop),
            ]
            self.redis_client.publish(self.__request_format.format('SubscribeMarketData'), json.dumps(inst_ids))
            rst = await asyncio.wait_for(cb, HANDLER_TIME_OUT, loop=self.io_loop)
            await sub_client.punsubscribe(channel_name1, channel_name2)
            sub_client.close()
            await asyncio.wait(tasks, loop=self.io_loop)
            return rst
        except Exception as e:
            logger.error('SubscribeMarketData failed: %s', repr(e), exc_info=True)
            if sub_client and sub_client.in_pubsub and channel_name1:
                await sub_client.unsubscribe(channel_name1, channel_name2)
                sub_client.close()
            return None

def UnSubscribeMarketData(self, inst_ids: list):
        sub_client = None
        channel_name1, channel_name2 = None, None
        try:
            sub_client = await aioredis.create_redis(
                (config.get('REDIS', 'host', fallback='localhost'),
                 config.getint('REDIS', 'port', fallback=6379)),
                db=config.getint('REDIS', 'db', fallback=1))
            channel_name1 = self.__market_response_format.format('OnRspUnSubMarketData', 0)
            channel_name2 = self.__market_response_format.format('OnRspError', 0)
            ch1, ch2 = await sub_client.psubscribe(channel_name1, channel_name2)
            cb = self.io_loop.create_future()
            tasks = [
                asyncio.ensure_future(self.query_reader(ch1, cb), loop=self.io_loop),
                asyncio.ensure_future(self.query_reader(ch2, cb), loop=self.io_loop),
            ]
            self.redis_client.publish(self.__request_format.format('UnSubscribeMarketData'), json.dumps(inst_ids))
            rst = await asyncio.wait_for(cb, HANDLER_TIME_OUT, loop=self.io_loop)
            await sub_client.punsubscribe(channel_name1, channel_name2)
            sub_client.close()
            await asyncio.wait(tasks, loop=self.io_loop)
            return rst
        except Exception as e:
            logger.error('SubscribeMarketData failed: %s', repr(e), exc_info=True)
            if sub_client and sub_client.in_pubsub and channel_name1:
                await sub_client.unsubscribe(channel_name1, channel_name2)
                sub_client.close()
            return None

def _initialize_redis_client(self):
        return await aioredis.create_redis(
            (config["redis"]["host"], config["redis"]["port"]),
            loop=asyncio.get_event_loop()
        )

def _create_connect():
    global _connection
    if not _connection:
        _connection = await aioredis.create_redis(
            (options.options.redis_host, options.options.redis_port),
            db=options.options.redis_index
        )
    return _connection

def Start(self):
        utils.prYellow("AIOREDIS")
        self.redis = await aioredis.create_redis((utils.secret["Redis"],6379),encoding='utf8')

def create_redis_client(_closable, loop, request):
    """Wrapper around aioredis.create_redis."""
    @asyncio.coroutine
    def f(*args, **kw):
        kw.setdefault('loop', loop)
        redis = yield from aioredis.create_redis(*args, **kw)
        _closable(redis)
        return redis
    return f

def test_order_filter_invalid(self):
        """
        Test invalid message order.
        """
        shark = SocketShark(TEST_CONFIG)
        await shark.prepare()
        client = MockClient(shark)
        session = client.session

        subscription = 'simple.topic'

        await session.on_client_event({
            'event': 'subscribe',
            'subscription': subscription,
        })
        assert client.log.pop() == {
            'event': 'subscribe',
            'subscription': subscription,
            'status': 'ok',
        }

        # Test message from server to client
        redis_settings = TEST_CONFIG['REDIS']
        redis = await aioredis.create_redis((
            redis_settings['host'], redis_settings['port']))
        redis_topic = redis_settings['channel_prefix'] + subscription

        await redis.publish_json(redis_topic, {
            'subscription': subscription,
            '_order': 'invalid',
            'data': {'foo': 'invalid'},
        })

        await redis.publish_json(redis_topic, {
            'subscription': subscription,
            'data': {'foo': 'bar'},
        })

        redis.close()

        # Wait for Redis to propagate the messages
        await asyncio.sleep(0.1)

        has_messages = await shark.run_service_receiver(once=True)
        assert has_messages

        assert client.log == [{
            'event': 'message',
            'subscription': subscription,
            'data': {'foo': 'bar'},
        }]

        await shark.shutdown()

def ReqOrderInsert(self, **kwargs):
        """
        InstrumentID ??
        VolumeTotalOriginal ??
        LimitPrice ??
        StopPrice ???
        Direction ??
        CombOffsetFlag ?,?,??
        ContingentCondition ????
        TimeCondition ????
        """
        sub_client = None
        channel_name1, channel_name2 = None, None
        try:
            sub_client = await aioredis.create_redis(
                (config.get('REDIS', 'host', fallback='localhost'),
                 config.getint('REDIS', 'port', fallback=6379)),
                db=config.getint('REDIS', 'db', fallback=1))
            request_id = self.next_id()
            order_ref = self.next_order_ref()
            kwargs['nRequestId'] = request_id
            kwargs['OrderRef'] = order_ref
            channel_name1 = self.__trade_response_format.format('OnRtnOrder', order_ref)
            channel_name2 = self.__trade_response_format.format('OnRspError', request_id)
            channel_name3 = self.__trade_response_format.format('OnRspOrderInsert', 0)
            ch1, ch2, ch3 = await sub_client.psubscribe(channel_name1, channel_name2, channel_name3)
            cb = self.io_loop.create_future()
            tasks = [
                asyncio.ensure_future(self.query_reader(ch1, cb), loop=self.io_loop),
                asyncio.ensure_future(self.query_reader(ch2, cb), loop=self.io_loop),
                asyncio.ensure_future(self.query_reader(ch3, cb), loop=self.io_loop),
            ]
            self.redis_client.publish(self.__request_format.format('ReqOrderInsert'), json.dumps(kwargs))
            rst = await asyncio.wait_for(cb, HANDLER_TIME_OUT, loop=self.io_loop)
            await sub_client.punsubscribe(channel_name1, channel_name2, channel_name3)
            sub_client.close()
            await asyncio.wait(tasks, loop=self.io_loop)
            logger.info('ReqOrderInsert, rst: %s', rst)
            return rst
        except Exception as e:
            logger.error('ReqOrderInsert failed: %s', repr(e), exc_info=True)
            if sub_client and sub_client.in_pubsub and channel_name1:
                await sub_client.unsubscribe(channel_name1, channel_name2)
                sub_client.close()
            return None

def init(loop):
    # Middlewares
    middlewares = [
        session_middleware(RedisStorage(await aioredis.create_pool((config.redis_ip, 6379)), cookie_name='w')),
        error_middleware,
        maintain_middleware,
    ]

    # init server
    app = web.Application(loop=loop,
                          middlewares=middlewares)

    redis = await aioredis.create_redis((config.redis_ip, config.redis['port']), loop=loop)
    app.redis = RedisFilter(redis)

    # Register admin account
    if await app.redis.get('User') is None:
        await app.redis.set('SecretKey', os.urandom(16), many=False)
        config.admin['password'] = await encrypt(app.redis, config.admin['password'])
        config.admin['permission'] = 0x0f
        await app.redis.set('User', config.admin, many=False)

    # Init Profile
    if await app.redis.get('Profile') is None:
        await app.redis.set('Profile', {
            'name': config.rss['author'],
            'link_desc': '',
            'text': ''
        }, many=False)

    # Security
    setup_security(app,
                   SessionIdentityPolicy(),
                   RedisAuthorizationPolicy(redis))

    await compass(app.router)
    CONST.CATEGORY = await app.redis.get('Categories') or []

    aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader(config.template_addr))

    _handler = app.make_handler(
        access_log=logger,
        access_log_format=formatters
    )
    _srv = await loop.create_server(_handler, config.server['host'], config.server['port'])
    print('Server started at http://%s:%s...' % (config.server['host'], config.server['port']))
    return _srv, _handler, app