Tasks

cronjob

cronjob(
    cron: str, /, on_startup: bool = False
) -> Callable[[TaskCallbackT], Includable[Cronjob]]

Run a task at the time specified by the cron schedule expression.

PARAMETER	DESCRIPTION
cron	The cronjob used to schedule when the callback is run. croniter is used for parsing cron expressions. TYPE: str
on_startup	If True, run the callback when this task is started. TYPE: bool DEFAULT: False

Loop

Loop(callback: TaskCallbackT, delay_seconds: float)

Bases: Task

_next_iteration

_next_iteration() -> float

Returns 0 if the first loop has not occurred, otherwise return Loop.delay_seconds.

set_interval

set_interval(
    *,
    hours: int = ...,
    minutes: int = ...,
    seconds: int = ...,
) -> None

set_interval(timedelta: _timedelta) -> None

set_interval(
    timedelta: _timedelta | None = None,
    *,
    hours: int = 0,
    minutes: int = 0,
    seconds: int = 0,
) -> None

Cancel the currently scheduled task and schedule the next task and future tasks with a new wait time.

Example

from datetime import datetime
import crescent
from crescent.ext import tasks

bot = hikari.GatewayBot("...")
client = crescent.Client(bot)

@client.include
@tasks.loop(seconds=1)
async def my_task():
    print(datetime.now())

@client.include
@crescent.command
async def set_interval(ctx: crescent.Context, interval: int):
    print(f"setting new interval to {interval}")
    my_task.metadata.set_interval(seconds=interval)
    await ctx.respond(f"Set new interval to {interval}s")

loop

loop(
    *,
    hours: int = ...,
    minutes: int = ...,
    seconds: int = ...,
) -> retT

loop(timedelta: _timedelta) -> retT

loop(
    timedelta: _timedelta | None = None,
    *,
    hours: int = 0,
    minutes: int = 0,
    seconds: int = 0,
) -> retT

Run a callback when the bot is started and every time the specified time interval has passed.

Task

Task(callback: TaskCallbackT)

Bases: ABC

_link staticmethod

_link(includable: Includable[_TaskType]) -> None

Sets hooks on Includable required for Task to function properly.

_next_iteration abstractmethod

_next_iteration() -> float

Returns how long to wait until the next iteration if a task was scheduled right now.