如何要求抽象方法是协程？

Question

如何要求抽象基 class 将特定方法实现为协程。例如，考虑这个 ABC：

import abc

class Foo(abc.ABC):
    @abc.abstractmethod
    async def func():
        pass

现在，当我子 class 并实例化时：

class Bar(Foo):
    def func():
        pass

b = Bar()

虽然 func 不是 ABC 中的 async，但它成功了。如果 func 是 async，我该怎么做才能成功？

Answer 1

您可以使用 __new__ 并检查 child class 是否以及如何覆盖 parent 的 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())

注意事项

a child class 也可以覆盖 __new__ 以抑制此约束

不仅可以等待 async。例如

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)

可以给change打电话

await o.change(key, value)

更不用说 objects 中的 __await__，其他原始 Futures、任务...

如何要求抽象方法是协程？

How require that an abstract method is a coroutine?

python

abstract-class

python-asyncio

注意事项