将* args和** kwargs传递给字典中的asyncio定期包装函数

Question

我正在使用asyncio从字典中收集任务并执行它们但我很难按预期工作。 （这是我的问题here的一个后续问题，但我重新编写了一些代码，因为它没有按预期工作，并决定使用包装函数代替它。）

所以我使用包装器来调用指定的任务函数。我希望包装器将任何* args或** kwargs转发到任务函数，并且如果设置了interval kwarg，还要在peridiocally重复任务。

如何将这些信息传递给包装器和任务函数，同时通过轻松地将新任务添加到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()

到目前为止，这似乎运作良好。但是，假设我希望Task-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. 如何以尽可能简单的方式向tasks字典中的特定函数提供* args和** kwargs（这样可以轻松添加新任务），并通过包装器将它们转发给任务函数本身？
3. 我的重复函数的方法是否定期纠正？我特别希望它在完成再次启动后才能完成睡眠，并且即使最后一个实例尚未完成，也不会再次启动。

Answer 1

使用functools.partial只有在实际包装faketask以包含可选关键字参数时才有意义。如果需要将关键字参数应用于其他函数（run_task），则需要单独执行此操作。例如，您可以在run_task dict中为tasks指定其他optoins：

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