Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

How require that an abstract method is a coroutine?

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?

like image 424
Björn Pollex Avatar asked Nov 29 '17 15:11

Björn Pollex


People also ask

What is Mutablemapping?

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.

What does import ABC do in Python?

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.

What is ABC collections?

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.

What is an abstract method in Python?

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.


1 Answers

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())

Caveats

  • a child class may override __new__ as well to suppress this constraint
  • not 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...

like image 74
kwarunek Avatar answered Nov 05 '22 08:11

kwarunek