coro-context-manager 0.2.0

A simple object to wrap coroutines to make them awaitable or used via an asyn context manager

coro-context-manager

coro-context-manager is a simple python package that includes an object that can wrap a coroutine to allow it to behave as a context manager or a regular awaitable.

This class is super useful when you have a coroutine that returns an object that defines an async context manager using __aenter__ and __aexit__

Installation

pip

pip install coro-context-manager

poetry

poetry add coro-context-manager

Usage

CoroContextManager can be used to wrap a coroutine so that it can be awaited or called via an async context manager in which case the library will try to use the underlying object's __aenter__ and __aexit__, if they exist.

import asyncio

from coro_context_manager import CoroContextManager


class MyObject:

    def __init__(self, initial_value):
        self.some_value = initial_value

    async def __aenter__(self):
        await asyncio.sleep(.1)
        self.some_value += 1
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await asyncio.sleep(.1)
        self.some_value -= 1

    @classmethod
    async def an_io_intensive_constructor(cls, initial_value):
        await asyncio.sleep(10)
        return cls(initial_value)


async def main():
    """
    Using CoroContextManager, I get a coroutine I can await or use with an async context manager, which proxies to
    the context manager defined on object returned by the coroutine, if it exists.
    """

    # i can await it directly
    myobj = await CoroContextManager(MyObject.an_io_intensive_constructor(5))
    print(type(myobj))
    # <class '__main__.MyObject'>

    # or use it as an async context manager, not having to await it, with the same api!
    async with CoroContextManager(MyObject.an_io_intensive_constructor(5)) as myobj:
        print(type(myobj))
        # <class '__main__.MyObject'>
        print(myobj.some_value)
        # 6

    print(myobj.some_value)
    # 5


asyncio.run(main())

Rationale

This is a common enough pattern used in several async packages all with slightly different implementation. It would be nice if there was a consistent pattern everyone was using; this package aims to provide that.

• aiopg
• aioodbc
• aiohttp
• cx_Oracle_async
• aiomysql

Latest Changes