запустить список методов класса в asyncio - PullRequest
Новая
       6

запустить список методов класса в asyncio

0 голосов
/ 06 марта 2020

Я пытаюсь использовать asyncio для запуска всех методов класса. Вот минимальный код: 

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)

Это запускает функции в операторе exe c во время выполнения фьючерсов на понимание списка. Какие есть альтернативы? Есть более 20 функций, и список продолжает расти. Следовательно, нужен компактный способ запуска всех методов класса с использованием asyncio.

Я использую python 3.6.8

1 Ответ

0 голосов
/ 22 марта 2020

Есть несколько способов извлечь весь объект publi c функции , я выбрал 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 publi c класса. 

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

PS

Он будет работать все класс publi c функции.

...