DEV Community

mutterings in the dark
mutterings in the dark

Posted on

Example of Asynchronous Context Manager (async with)

#!/usr/bin/env python3
import asyncio

async def sleep(time):
    print(f"Sleeping for {time} seconds")
    await asyncio.sleep(time)

class MyObj:
    def __init__(self, warn=True):
        """
        Don't directly create an instance
        Instead use async_init() below (to work-around async call in __init__)

        Created with help from Stack Overflow:
            How to set class attribute with await in __init__
            https://stackoverflow.com/q/33128325
            Asked by https://stackoverflow.com/users/1026990/uralbash
            https://stackoverflow.com/a/33134213
            Answered by https://stackoverflow.com/users/2073595/dano
        """
        if warn:
            assert False, f"Do not directly create an instance of {type(self)}"

    # pylint: disable=attribute-defined-outside-init
    @classmethod
    async def async_init(cls, time):
        obj = cls(warn=False)
        obj.time = time
        await sleep(obj.time)
        return obj

    async def __aenter__(self):
        print("Enter")
        await sleep(self.time)
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        print("Exit")

async def main():
    async with await MyObj.async_init(1) as my_obj:
        pass
Enter fullscreen mode Exit fullscreen mode

Alternatively with decorator asynccontextmanager and try -> yield -> finally

#!/usr/bin/env python3
import asyncio
from contextlib import asynccontextmanager

async def sleep(time):
    print(f"Sleeping for {time} seconds")
    await asyncio.sleep(time)

class MyObj:

    @classmethod
    @asynccontextmanager
    async def async_init(cls, time):
        obj = cls()
        obj.time = time
        await sleep(obj.time)
        try:
            print("Enter")
            await sleep(obj.time)
            yield obj
        except AssertionError:
            print("Assertion caught")
        finally:
            print("Exit")

async def main():
    async with MyObj.async_init(1) as my_obj:
        assert False

if __name__ == "__main__":
    asyncio.run(main())
Enter fullscreen mode Exit fullscreen mode

Top comments (0)