asyncinit module
This package provides the asyncinit
decorator, which enables an asynchronous constructor
to be called like any other asynchronous function.
Example
from asyncinit import asyncinit @asyncinit class MyClass: async def __init__(self, param): self.val = await self.deferredFn(param) async def deferredFn(self, x): # ... return x + 2 obj = await MyClass(42) assert obj.val == 44
Inheritance
Note that if a parent class is decorated with @asyncinit
, its child classes
automatically inherit async __init__
functionality and don't require decoration
themselves (as long as they don't override __new__
.)
""" This package provides the `asyncinit` decorator, which enables an asynchronous constructor to be called like any other asynchronous function. ## Example ```python3 from asyncinit import asyncinit @asyncinit class MyClass: async def __init__(self, param): self.val = await self.deferredFn(param) async def deferredFn(self, x): # ... return x + 2 obj = await MyClass(42) assert obj.val == 44 ``` ## Inheritance Note that if a parent class is decorated with `@asyncinit`, its child classes automatically inherit async `__init__` functionality and don't require decoration themselves (as long as they don't override `__new__`.) """ import functools import inspect def asyncinit(obj): """ Add async `__init__` functionality to the given class. """ if not inspect.isclass(obj): raise ValueError("decorated object must be a class") if obj.__new__ is object.__new__: cls_new = _new else: cls_new = _force_async(obj.__new__) @functools.wraps(obj.__new__) async def new(cls, *args, **kwargs): self = await cls_new(cls, *args, **kwargs) cls_init = _force_async(self.__init__) await cls_init(*args, **kwargs) return self obj.__new__ = new return obj # Force the given function to be `await`-able. def _force_async(fn): if inspect.iscoroutinefunction(fn): return fn async def wrapped(*args, **kwargs): return fn(*args, **kwargs) return wrapped # Wraps `object.__new__` in a coroutine, only passing it the class object. This kludge is # required because that function throws `TypeError: object() takes no parameters` if # passed any other parameters. async def _new(cls, *args, **kwargs): return object.__new__(cls) def _test_asyncinit(): import asyncio async def deferredFn(z): return z * 4 @asyncinit class TestAsyncInit: async def __init__(self, x): self.num = x self.deferred = await deferredFn(x) @asyncinit class TestNormalInit: def __init__(self, x): self.num = x self.derived = 4 * x @asyncinit class SuperClass: async def __init__(self, a, b, c): self.a = a self.b = b self.c = c class SubClass(SuperClass): async def __init__(self, a): await super().__init__(a, 2 * a, 3 * a) async def mainTask(): test = await TestAsyncInit(3) assert test.num == 3 assert test.deferred == 12 test = await TestNormalInit(42) assert test.num == 42 assert test.derived == 168 test = await SubClass(4) assert test.a == 4 assert test.b == 8 assert test.c == 12 eventLoop = asyncio.get_event_loop() eventLoop.run_until_complete(mainTask())
Functions
def asyncinit(
obj)
Add async __init__
functionality to the given class.
def asyncinit(obj): """ Add async `__init__` functionality to the given class. """ if not inspect.isclass(obj): raise ValueError("decorated object must be a class") if obj.__new__ is object.__new__: cls_new = _new else: cls_new = _force_async(obj.__new__) @functools.wraps(obj.__new__) async def new(cls, *args, **kwargs): self = await cls_new(cls, *args, **kwargs) cls_init = _force_async(self.__init__) await cls_init(*args, **kwargs) return self obj.__new__ = new return obj