Class AsyncClass

Is a base wrapper with metaclass has no TaskStore instance and additional methods like self.create_task and self.create_future.

This class just solves the initialization problem:

import asyncio
from async_class import AsyncClass


class MyAsyncClass(AsyncClass):
   async def __ainit__(self):
       future = self.loop.create_future()
       self.loop.call_soon(future.set_result, True)
       await future


async def main():
   instance = await MyAsyncClass()
   print(instance)

asyncio.run(main())

Class AsyncObject

Base class with task store instance and helpers for simple task management.

import asyncio
from async_class import AsyncObject


class MyClass(AsyncObject):
   async def __ainit__(self):
       self.task = self.create_task(asyncio.sleep(3600))


async def main():
   obj = await MyClass()

   assert not obj.task.done()

   await obj.close()

   assert obj.task.done()


asyncio.run(main())

Class TaskStore

TaskStore is a task management helper. One instance has create_task() and create_future() methods and all created entities will be destroyed when TaskStore will be closed via close() method.

Also, a task store might create a linked copy of the self, which will be closed when the parent instance will be closed.

import asyncio
from async_class import TaskStore


async def main():
   store = TaskStore(asyncio.get_event_loop())

   task1 = store.create_task(asyncio.sleep(3600))

   child_store = store.get_child()
   task2 = child_store.create_task(asyncio.sleep(3600))

   await store.close()

   assert task1.done() and task2.done()


asyncio.run(main())