Run aioredis with trio_asyncio

test_aioredis.py

import asyncio
from contextlib import asynccontextmanager, AsyncExitStack

import aioredis
import trio_asyncio


async def test_aioredis():
    """ test to use aioredis in trio using trio_asyncio """
    async with trio_asyncio.open_loop():
        fut_conn = asyncio.ensure_future(aioredis.create_redis('redis://127.0.0.1', encoding="utf-8"))
        redis = await trio_asyncio.run_coroutine(fut_conn)

        fut_set = asyncio.ensure_future(redis.set('my-key', 'value'))
        await trio_asyncio.run_coroutine(fut_set)

        fut_get = asyncio.ensure_future(redis.get('my-key'))
        val = await trio_asyncio.run_coroutine(fut_get)

        assert val == "value"
        redis.close()

        fut_wait_closed = asyncio.ensure_future(redis.wait_closed())
        await trio_asyncio.run_coroutine(fut_wait_closed)


async def test_aioredis_connection_pool():
    """ test to create connection pool """
    async with trio_asyncio.open_loop() as loop:
        fut_conn = asyncio.ensure_future(aioredis.create_redis_pool(
            'redis://127.0.0.1', encoding="utf-8",
            minsize=5, maxsize=10,
            loop=loop))
        redis = await trio_asyncio.run_coroutine(fut_conn)

        fut_set = asyncio.ensure_future(redis.set('my-key', 'value'))
        await trio_asyncio.run_coroutine(fut_set)

        fut_get = asyncio.ensure_future(redis.get('my-key'))
        val = await trio_asyncio.run_coroutine(fut_get)

        assert val == "value"
        redis.close()

        fut_wait_closed = asyncio.ensure_future(redis.wait_closed())
        await trio_asyncio.run_coroutine(fut_wait_closed)


@asynccontextmanager
async def create_connection():
    fut_conn = asyncio.ensure_future(aioredis.create_redis('redis://127.0.0.1', encoding="utf-8"))
    redis = await trio_asyncio.run_coroutine(fut_conn)
    try:
        yield redis
    finally:
        redis.close()
        fut_wait_closed = asyncio.ensure_future(redis.wait_closed())
        await trio_asyncio.run_coroutine(fut_wait_closed)


async def test_aioredis_create_connection():
    """ test to create connection using asynccontextmanager """
    async with trio_asyncio.open_loop():
        async with AsyncExitStack() as stack:
            redis = await stack.enter_async_context(create_connection())

            fut_set = asyncio.ensure_future(redis.set('my-key', 'value'))
            await trio_asyncio.run_coroutine(fut_set)

            fut_get = asyncio.ensure_future(redis.get('my-key'))
            val = await trio_asyncio.run_coroutine(fut_get)
            assert val == "value"


@trio_asyncio.aio_as_trio
async def set_and_get_value(redis: aioredis.Redis):
    await redis.set('my-key', 'value')
    return await redis.get('my-key')


async def test_aioredis_decorator():
    """ test to use aio_as_trio decorator """
    async with trio_asyncio.open_loop():
        async with AsyncExitStack() as stack:
            redis = await stack.enter_async_context(create_connection())
            val = await set_and_get_value(redis)
            assert val == "value"


@asynccontextmanager
async def create_connection_aio():
    redis = await aioredis.create_redis('redis://127.0.0.1', encoding="utf-8")
    try:
        yield redis
    finally:
        redis.close()
        await redis.wait_closed()


async def test_aioredis_connection_aio():
    """ test to wrap asyncio generator by aio_as_trio

    See: https://github.com/python-trio/trio-asyncio/blob/master/trio_asyncio/base.py
    """
    async with trio_asyncio.open_loop():
        async with AsyncExitStack() as stack:
            redis = await stack.enter_async_context(trio_asyncio.aio_as_trio(create_connection_aio()))
            val = await set_and_get_value(redis)
            assert val == "value"