Python：用变量字符串的名称定义函数答案

【问题标题】：Python: Define function with name from variable stringPython：用变量字符串的名称定义函数
【发布时间】：2021-08-02 00:33:45
【问题描述】：

我正在构建一个具有多个任务循环且间隔不同的脚本。自然地，我想制作一个 for 循环来定义所有这些，以使其占用更少的空间。但是，似乎不可能这样做。

我怎样才能缩短这个sn-p？真的有办法吗？

timeloops = ["60","600","3600","7200","14400","21600"]

@tasks.loop(seconds=60)
async def task_60(self):
    await second_func(self,channels["60"])

@tasks.loop(seconds=600)
async def task_600(self):
    await second_func(self,channels["600"])

@tasks.loop(seconds=3600)
async def task_3600(self):
    await second_func(self,channels["3600"])

@tasks.loop(seconds=7200)
async def task_7200(self):
    await second_func(self,channels["7200"])

@tasks.loop(seconds=14400)
async def task_14400(self):
    await second_func(self,channels["14400"])

@tasks.loop(seconds=21600)
async def task_21600(self):
    await second_func(self,channels["21600"])

这里的另一个问题导致我使用 Globals，但似乎这只是用于调用该函数，而不是用于定义它。

提前谢谢你。

【问题讨论】：

这可能对stackoverflow.com/questions/11553721/…有帮助。

标签： python function loops

【解决方案1】：

由于您没有提供 MRE，我无法对此进行测试，但我认为将它们构建为列表应该是可行的：

task_loops = [
    tasks.loop(seconds=int(secs))(
        lambda self, secs=secs: (
            await second_func(self.channels[secs]) for _ in '_'
        ).__anext()
    ) for secs in timeloops
]

我有点模糊的部分是您是否可以这样定义异步 lambda。这也应该有效：

task_loops = []
for secs in timeloops:
    @tasks.loop(seconds=int(secs))
    async def task_secs(self, secs=secs):
        await second_func(self, channels[secs])
    task_loops.append(task_secs)

【讨论】：

【解决方案2】：

如果你真的需要它们作为你的类的方法，那么你可以通过创建你的任务方法的参数化版本来让它工作：

async def timed_task(self, t):
    await second_func(self, channels[t])

然后使用functools.partialmethod函数创建方法的各个实例（也许把这个放在类的__init__中）：

from functools import partialmethod

for t in timeloops:
    # Create the tasks.loop wrapper
    wrapper = tasks.loop(seconds=int(t))
    # Create the method to be wrapped
    wrapped = partialmethod(timed_task, t)
    setattr(self, f"task_{t}", wrapper(wrapped))

【讨论】：

【解决方案3】：

您可以使用exec 执行函数定义并使用字符串格式填充变量。你可以试试下面的代码。

import inspect

timeloops = ["60", "600", "3600", "7200", "14400", "21600"]

for i in timeloops:
    define_func = f"""
    @tasks.loop(seconds=int(i))
    async def task_{i}(self):
        await second_func(self,channels[f"{i}"])
    """
    define_func = inspect.cleandoc(define_func)
    exec(define_func)

或者更简单的方法

timeloops = ["60", "600", "3600", "7200", "14400", "21600"]

for i in timeloops:
    @tasks.loop(seconds=int(i))
    async def task(self, i=i):  # Prevent variable i from being overwritten
        await second_func(self,channels[i])

    exec(f'task_{i}=task')

【讨论】：