在asyncio中运行类方法列表

有几种方法可以 extract all the object public functions ,我选择了 dir 内置功能。

一旦你拥有了所有的功能,你就可以使用 inspect.iscoroutinefunction 为了知道函数是否为 coroutine 作用

请考虑以下几点 class Reports 例子:

class Reports:
   def __init__(self, name):
      self.name = name

   async def __class_private(self):
       print('__class_private')

   async def _private(self):
       print('_private')

   async def func1(self, *args, **kwargs):
        print('func 1')

   async def func2(self, *args, **kwargs):
       print('func 2')

   def func3(self, *args, **kwargs):
       print('func 3')

下面的代码将提取所有 async 平民的 课程的功能。

    report = Reports('dealer1')
    func_list  = [getattr(report, func) for func in dir(report) if callable(getattr(report, func)) and not func.startswith('_')]

它一次处理所有类函数并排除所有私有函数。

现在是第二部分,运行所有这些(你可以使用列表理解来完成,我想展示流程和输出):

    for f in func_list:
        if inspect.iscoroutinefunction(f):
            print(f.__name__)
            tasks.append(loop.create_task(f()))
    print('Results')
    done, pending = await asyncio.wait(tasks)
    print(done)
    print(pending)

运行它将产生以下结果:

func1
func2
Results
func 1
func 2
{<Task finished coro=<Reports.func2() done, defined at C:/stackoverflow/tests:18> result=None>, <Task finished coro=<Reports.func1() done, defined at C:/stackoverflow/tests:15> result=None>}
set()

附笔

它将运行 全部的 班级 平民的 功能。