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.

1

3 Answers 3

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)
Sign up to request clarification or add additional context in comments.

Comments

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

Comments

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

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.