如何在指定* args或** kwargs的同时从函数字典运行函数 [英] How to run functions from a dict of functions while specifying *args or **kwargs

问题描述

我基本上想运行从功能字典收集的任务。我在程序中使用了asyncio，但这对我的问题可能并不重要。这是我的代码来说明...

I basically want to run tasks that are gathered from a dict of functions. I'm using asyncio in my program but it probably doesn't matter for my question. Here is my code to illustrate...

import asyncio
import random


async def faketask(taskname):    # This is a function to simulate asynchronous tasks being performed.
    fakedelay = random.randint(1,6)
    print(f'{taskname} started (completing in {fakedelay} seconds)')
    await asyncio.sleep(fakedelay)
    print(f'{taskname} completed after {fakedelay} seconds')


async def main():
    tasklist = {    # This is a dict of tasks to be performed
    'Task-1': faketask,
    'Task-2': faketask,
    'Task-3': faketask,
    }

    _tasks = []
    for taskname in tasklist:
        _tasks.append(asyncio.ensure_future(tasklist[taskname](taskname)))
        print(f'Added {taskname} to job queue.')
    await asyncio.gather(*_tasks)


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

这很好用。但是我想定期重复执行某些任务，因此我向 faketask 函数添加了一些代码。

This works pretty well. But I want to repeat certain tasks periodically, so I added some code to the faketask function.

async def faketask(taskname, interval=None):
    fakedelay = random.randint(1,6)
    print(f'{taskname} started (completing in {fakedelay} seconds)')
    await asyncio.sleep(fakedelay)
    print(f'{taskname} completed after {fakedelay} seconds')
    if interval is not None:
        print(f'Repeating {taskname} in {interval} seconds')
        while True:
            await faketask(taskname, *args, **kwargs)
            await asyncio.sleep(interval)

这通过向kwarg interval 提供一个整数。将来的功能可能还会使用其他* args和** kwargs。

This makes the function repeat by providing the kwarg interval with an integer. Future functions may also use additional *args and **kwargs.

所以我基本上希望我可以在任务列表字典中指定重复间隔，例如

So I basically wish I could specify the repeat interval in the tasklist dict, e.g.

    tasklist = {
    'Task-1': faketask,
    'Task-2': faketask,
    'Task-3': faketask(interval=60),
    }

但这不是无法正常工作，因为 faketask（）缺少1个必需的位置参数：'taskname'。

But that doesn't work because faketask() is missing 1 required positional argument: 'taskname'.

有什么聪明的方法可以解决这个问题？

Is there any clever way to solve this?

作为奖励问题， _tasks.append（asyncio.ensure_future（tasklist [taskname]（taskname）））看起来有点难看，有什么方法可以自动传递 taskname 参数吗？

And as a bonus question, the line _tasks.append(asyncio.ensure_future(tasklist[taskname](taskname))) looks a bit ugly, is there any way to pass the taskname argument automatically?

推荐答案

您可以使用 functools.partial ，它专门用于以下情况：

You can use functools.partial, which is designed exactly for situations like this:

'Task-3': functools.partial(faketask, interval=60),

作为奖励问题，行 _tasks.append（asyncio.ensure_future（tasklist [taskname]（taskname））） 看起来有点丑陋，有没有办法自动传递taskname参数？

And as a bonus question, the line _tasks.append(asyncio.ensure_future(tasklist[taskname](taskname))) looks a bit ugly, is there any way to pass the taskname argument automatically?

您可以使用 items（）删除冗余：

tasks = []
for taskname, taskfn in tasklist.items():
    tasks.append(taskfn(taskname))

不需要显式调用 asyncio.ensure_future ，因为 asyncio.gather 会自动执行此操作。另外，我将 _tasks 重命名为 tasks 。按照惯例，变量名前面的 _ 表示该变量将不会在后续代码中使用，在这里不是这种情况。

The explicit call to asyncio.ensure_future is unnecessary because asyncio.gather does that automatically. Also, I've renamed _tasks to just tasks. By convention, the _ in front of a variable name denotes that the variable will not be used in subsequent code, which is not the case here.

这篇关于如何在指定* args或** kwargs的同时从函数字典运行函数的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！