How can I require that an abstract base class implement a specific method as a coroutine. For example, consider this ABC:
import abc
class Foo(abc.ABC):
@abc.abstractmethod
async def func():
pass
Now when I subclass and instantiate that:
class Bar(Foo):
def func():
pass
b = Bar()
This succeeds, although func
is not async
, as in the ABC. What can I do so that this only succeeds if func
is async
?
1.0. interface MutableMap<K, V> : Map<K, V> A modifiable collection that holds pairs of objects (keys and values) and supports efficiently retrieving the value corresponding to each key. Map keys are unique; the map holds only one value for each key.
Python has a module called abc (abstract base class) that offers the necessary tools for crafting an abstract base class. First and foremost, you should understand the ABCMeta metaclass provided by the abstract base class. The rule is every abstract class must use ABCMeta metaclass.
ABC Collectors is known as a debt collections services company. Established in 1964, they have been in business for approximately 55 years, and have plenty of years under their corporation heading.
An abstract method is a method that is declared, but contains no implementation. Abstract classes cannot be instantiated, and require subclasses to provide implementations for the abstract methods.
You may use __new__
and check if and how a child class has override parent's coros.
import asyncio
import abc
import inspect
class A:
def __new__(cls, *arg, **kwargs):
# get all coros of A
parent_coros = inspect.getmembers(A, predicate=inspect.iscoroutinefunction)
# check if parent's coros are still coros in a child
for coro in parent_coros:
child_method = getattr(cls, coro[0])
if not inspect.iscoroutinefunction(child_method):
raise RuntimeError('The method %s must be a coroutine' % (child_method,))
return super(A, cls).__new__(cls, *arg, **kwargs)
@abc.abstractmethod
async def my_func(self):
pass
class B(A):
async def my_func(self):
await asyncio.sleep(1)
print('bb')
class C(A):
def my_func(self):
print('cc')
async def main():
b = B()
await b.my_func()
c = C() # this will trigger the RuntimeError
await c.my_func()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
__new__
as well to suppress this constraintnot only async
may be awaited. For example
async def _change_in_db(self, key, value):
# some db logic
pass
def change(self, key, value):
if self.is_validate(value):
raise Exception('Value is not valid')
return self._change_in_db(key, value)
it's ok to call change
like
await o.change(key, value)
Not to mention __await__
in objects, other raw Futures, Tasks...
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With