0

I'm building a script which has multiple task loops, with different intervals. Naturally I would want to make a for-loop to define all of these, to make it take up less space. However, it seems like it's not possible to do so.

How can I shorten down this snippet? Is there actually a way?

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

Another question on here lead me to use Globals, but it seems like that's only for calling the function, and not for defining it.

Thank you in advance.

Fox
  • 32
  • 6

3 Answers3

1

I'm not able to test this since you didn't provide an MRE, but I think building them as a list should be doable:

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

The part I'm a little fuzzy on is whether you can define an async lambda that way. This should also work:

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)
Samwise
  • 68,105
  • 3
  • 30
  • 44
1

If you really need them as methods on your class then you may be able to get it working by creating a parameterised version of your tasks method:

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

And then using the functools.partialmethod function to create the individual instances of the method (perhaps put this in the __init__ of the class):

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))
Malcolm
  • 315
  • 2
  • 9
1

You can use exec to execute function definition and use string formatting to fill variables. You can try the following code.

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)

or the simpler way

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')
fitz
  • 540
  • 4
  • 11