Передача * args и ** kwargs в асинхронную периодическую функцию-обертку из словаря - PullRequest
Новая
       14

Передача * args и ** kwargs в асинхронную периодическую функцию-обертку из словаря

0 голосов
/ 08 апреля 2019

Я использую asyncio для сбора заданий из словаря и их выполнения, но у меня возникают трудности с тем, чтобы он работал так, как задумано. (Это своего рода дополнительный вопрос к моему вопросу здесь , но я немного переписал код, потому что он не работал должным образом, и решил, что будет лучше использовать оболочкувместо этого.)

Поэтому я использую оболочку для вызова указанной функции задачи.Я хочу, чтобы оболочка перенаправляла любые * args или ** kwargs в функцию задачи, а также повторяла задачу периодически, если установлен interval kwarg.

Как передать эту информацию оболочке ифункция задачи, сохраняя ее легко обслуживаемой с возможностью легко добавлять новые задачи в словарь tasks?

Пожалуйста, взгляните на мой код для иллюстрации. 

import asyncio
import random

async def run_task(taskname, taskfunc, interval=None, *args, **kwargs):
    # Wrapper which will run the specified function, and repeat it if 'interval' is set.
    # Should also be able to pass any potential *args and **kwargs to the function.
    fakedelay = random.randint(1,6)
    print(f'{taskname} started (completing in {fakedelay} seconds)')
    await taskfunc(fakedelay, *args, **kwargs)
    print(f'{taskname} completed after {fakedelay} seconds')
    if interval is not None:
        print(f'Repeating {taskname} in {interval} seconds...')
        while True:
            await taskfunc(fakedelay, *args, **kwargs)
            await asyncio.sleep(interval)

async def faketask(fakedelay, *args, **kwargs):
    # Function to simulate a coroutine task
    await asyncio.sleep(fakedelay)

async def main():
    tasks = {
        # Dictionary of tasks to perform
        'Task-1': faketask,
        'Task-2': faketask,
        'Task-3': faketask,
    }

    tasklist = []
    for taskname, taskfunc in tasks.items():
        tasklist.append(run_task(taskname, taskfunc))
        print(f'Added {taskname} to job queue.')
    await asyncio.gather(*tasklist)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

Это кажетсядо сих пор хорошо работать.Но допустим, что я хочу, чтобы Задача-3 повторялась каждые 10 секунд после каждого ее завершения. Я хотел бы просто указать ее в словаре tasks, чтобы было как можно проще добавлять новыезадачи на будущее.Например: 

tasks = {
    # Dictionary of tasks to perform
    'Task-1': faketask,
    'Task-2': faketask,
    'Task-3': faketask(interval=10),
}

Но выполнение этого дает TypeError: faketask() missing 1 required positional argument: 'fakedelay' Полагаю, это имеет смысл, потому что interval kwarg предназначен для оболочки, а не для самой функции задачи (faketask).И оболочка, кажется, не в состоянии добавить * args или ** kwargs (fakedelay в этой ситуации).

В моем предыдущем вопросе мне предложили использовать functools.partial. 

tasks = {
    'Task-1': faketask,
    'Task-2': faketask,
    'Task-3': functools.partial(faketask, interval=10),
}

Это несколько решило проблему из моего предыдущего вопроса, но после переписывания кода и добавления функции-обертки это, похоже, сейчас ничего не делает, и по общему признанию, мне трудно понять, как functools.partial должен бытьб.

Итак, мои вопросы:

  1. Как я могу это сделать, это подходящий способ для достижения того, что я пытаюсь сделать?

  2. Как я могу предоставить * args и ** kwargs для определенной функции в словаре tasks настолько простым, насколько это возможно (чтобы можно было легко добавлять новые задачи), и иметьони перенаправляются в саму задачу через оболочку?

  3. Правильно ли периодически повторять мой метод повторения функции?Я специально хочу, чтобы он только спал после завершения перед повторным запуском, а не просто снова запускался, даже если последний экземпляр еще не закончился.

1 Ответ

2 голосов
/ 08 апреля 2019

Использование functools.partial имеет смысл, только если вы на самом деле переносите faketask, чтобы включить необязательный аргумент ключевого слова.Если вам нужно применить ключевое слово аргумент к другой функции (run_task), то вам нужно сделать это независимо.Например, вы можете указать дополнительные опции для run_task в tasks dict: 

tasks = {
    'Task-1': faketask,
    'Task-2': faketask,
    'Task-3': (faketask, {'interval': 10)),
}

Код, который вызывает run_task, должен будет затем распознавать кортежи: 

for taskname, taskfunc_maybe_with_options in tasks.items():
    if isinstance(taskfunc_maybe_with_options, tuple):
        taskfunc, options = taskfunc_maybe_with_options
    else:
        taskfunc = taskfunc_maybe_with_options
        options = {}
    tasklist.append(run_task(taskname, taskfunc, **options))
...