run a list of class methods in asyncio

Question

I am trying to use asyncio to run all the methods of a class. Here is a minimal code:

Class Reports:
   def __init__(self, name):
      self.name = name
   async def func1(**kwargs):
        await subfunc1
   async def func2(**kwargs):
        await subfunc2

report = Reports('dealer1)
func_list = ['func1', 'func2']
keyworded_args = {'arg1':arg1, 'arg2' : arg2}
futures = [exec('reports' + func(**keyworded_args)) for func in func_list]
loop = asyncio.get_event_loop()
loop.run_until_complete(futures)

This runs the functions in the exec statement during executing the list comprehension futures. What are the alternatives? There are more than 20 functions and the list keeps on increasing. Hence want a compact way to run all the methods of the class using asyncio.

I use python 3.6.8

score 0 · Answer 1 · answered Mar 22 '20 at 14:48

There are several ways you can extract all the object public functions, I chose dir built-in function.

Once you have all the functions you can use the inspect.iscoroutinefunction in order to know if the function is a coroutine function.

Please consider the following class Reports example:

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

The following code will extract all the async public function of the class.

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

It's taking all the class function and exclude all the private once.

Now for the second part, running all of them (you can do it using list comprehension, I wanted to show the process and outputs):

    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)

Running it will results with:

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

P.S

It will run all the class public functions.

run a list of class methods in asyncio

1 Answers1