disagreement/docs/task_loop.md

Task Loops

The tasks extension allows you to run functions periodically. Decorate an async function with @tasks.loop and start it using .start().

from disagreement.ext import tasks

@tasks.loop(minutes=1.0)
async def announce():
    print("Hello from a loop")

announce.start()

Stop the loop with .stop() when you no longer need it.

You can provide the interval in seconds, minutes, hours or as a datetime.timedelta:

import datetime

@tasks.loop(delta=datetime.timedelta(seconds=30))
async def ping():
    ...

Handle exceptions raised by the looped coroutine using on_error:

async def log_error(exc: Exception) -> None:
    print("Loop failed:", exc)

@tasks.loop(seconds=5.0, on_error=log_error)
async def worker():
    ...

Run setup and teardown code using before_loop and after_loop:

@tasks.loop(seconds=5.0)
async def worker():
    ...

@worker.before_loop
async def before_worker():
    print("starting")

@worker.after_loop
async def after_worker():
    print("stopped")

You can also schedule a task at a specific time of day:

from datetime import datetime, timedelta

time_to_run = (datetime.now() + timedelta(seconds=5)).time()

@tasks.loop(time_of_day=time_to_run)
async def daily_task():
    ...