模块 'aioredis' 没有属性 'create_redis'

Question

使用 python 3.6.12 和 aioredis 2.0.0，asyncio 3.4.3

尝试使用 aioredis 中的代码片段进行测试 pub/sub:

import asyncio
import aioredis


async def reader(ch):
    while (await ch.wait_message()):
        msg = await ch.get_json()
        print("Got Message:", msg)


async def main():
    pub = await aioredis.create_redis(
        'redis://:password@localhost:6379')
    sub = await aioredis.create_redis(
        'redis://:password@localhost:6379')
    res = await sub.subscribe('chan:1')
    ch1 = res[0]

    tsk = asyncio.ensure_future(reader(ch1))

    res = await pub.publish_json('chan:1', ["Hello", "world"])
    assert res == 1

    await sub.unsubscribe('chan:1')
    await tsk
    sub.close()
    pub.close()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())

但是不断弹出以下错误。

Traceback (most recent call last):
  File "tests/test_async_redis.py", line 32, in <module>
    result = loop.run_until_complete(main())
  File "/Users/dustinlee/.pyenv/versions/3.6.12/lib/python3.6/asyncio/base_events.py", line 488, in run_until_complete
    return future.result()
  File "tests/test_async_redis.py", line 12, in main
    pub = await aioredis.create_redis(
AttributeError: module 'aioredis' has no attribute 'create_redis'

谁能告诉我我做错了什么？可能是显而易见的事情，但我只是没有看到。谢谢！

Answer 1

aioredis 从版本 2.0 开始，现在遵循库 redis-py.[=22 的 public API 实现=]

来自aioredis doc page

aioredis v2.0 is now a completely compliant asyncio-native implementation of redis-py. The entire core and public API has been re-written to follow redis-py‘s implementation as closely as possible.

所以方法aioredis.create_redis不再是public API你可以在2.0版本中用来建立连接。如果您希望 create_redis 方法起作用，请使用低于 2 的版本。

你可以参考新的pub sub example。

复制到这里的代码以防 link 以后不起作用。

import asyncio
import async_timeout
import aioredis

STOPWORD = "STOP"


async def pubsub():
    redis = aioredis.Redis.from_url(
        "redis://localhost", max_connections=10, decode_responses=True
    )
    psub = redis.pubsub()

    async def reader(channel: aioredis.client.PubSub):
        while True:
            try:
                async with async_timeout.timeout(1):
                    message = await channel.get_message(ignore_subscribe_messages=True)
                    if message is not None:
                        print(f"(Reader) Message Received: {message}")
                        if message["data"] == STOPWORD:
                            print("(Reader) STOP")
                            break
                    await asyncio.sleep(0.01)
            except asyncio.TimeoutError:
                pass

    async with psub as p:
        await p.subscribe("channel:1")
        await reader(p)  # wait for reader to complete
        await p.unsubscribe("channel:1")

    # closing all open connections
    await psub.close()


async def main():
    tsk = asyncio.create_task(pubsub())

    async def publish():
        pub = aioredis.Redis.from_url("redis://localhost", decode_responses=True)
        while not tsk.done():
            # wait for clients to subscribe
            while True:
                subs = dict(await pub.pubsub_numsub("channel:1"))
                if subs["channel:1"] == 1:
                    break
                await asyncio.sleep(0)
            # publish some messages
            for msg in ["one", "two", "three"]:
                print(f"(Publisher) Publishing Message: {msg}")
                await pub.publish("channel:1", msg)
            # send stop word
            await pub.publish("channel:1", STOPWORD)
        await pub.close()

    await publish()


if __name__ == "__main__":
    import os

    if "redis_version:2.6" not in os.environ.get("REDIS_VERSION", ""):
        asyncio.run(main())

您也可以参考 redis-py docs，因为它应该是 aioredis 2.0 现在紧随其后的内容。