lightbulb.tasks¶

This module implements simple asynchronous tasks that run alongside the Lightbulb client.

Usage Guide¶

Creating Tasks¶

Tasks can be created using one of the provided decorators:

When creating a task, you must pass a trigger callable that will be used to know when and how often the task should be invoked.

Built-in Triggers¶

Lightbulb has two built-in triggers that you can use for your own tasks. These are uniformtrigger and crontrigger.

uniformtrigger

When used, this trigger will invoke the task at the given uniform interval. I.e. if you set the interval to 10 seconds, the task will be invoked every 10 seconds. The countdown before each invocation will start once the previous invocation has finished.

crontrigger

Important

To use this trigger you need to install Lightbulb with the optional dependency [crontrigger]

When used, this trigger invokes the task following the given crontab. For example, you could use this trigger to schedule the task to run every day at 10am. This trigger’s schedule is always calculated from UTC.

Custom Triggers¶

Creating your own custom trigger is easy. A trigger is any callable that takes a single parameter - this will be an instance of TaskExecutionData - and returns a float. This will be called after every invocation of the task in order to calculate how much time to wait until the next task invocation.

The task execution data object contains some information you can use if you wish to set next the interval based on when the task was last invoked, or how long the last invocation took.

Running Tasks¶

Tasks registered to the client will be run automatically once the client’s start method has been called. Any tasks registered after the client has already been started will be run immediately. If, when creating your task, you pass auto_start=False, then you will instead need to call the task’s start method manually.

If the task is registered to a loader, once the loader is removed from the client the task will be cancelled.

class Task(func: Callable[[...], Awaitable[Any]], trigger: Callable[[TaskExecutionData], float | Coroutine[Any, Any, float]], auto_start: bool, max_failures: int, max_invocations: int)[source]¶

Class representing an asynchronous repeating task.

Parameters:

func – The function to execute. Dependency injection will be enabled for it once the task is created.
trigger – The trigger function to use to resolve the interval between task executions.
auto_start – Whether the task should be started automatically. This means that if the task is added to the client upon the client being started, the task will also be started; it will also be started if being added to an already-started client.
max_failures – The maximum number of failed attempts to execute the task before it is cancelled. Setting this to a negative number will prevent the task from being cancelled, regardless of how often the task fails.
max_invocations – The maximum number of times the task can be invoked before being stopped. Setting this to a negative number will disable this behaviour, allowing unlimited invocations.

async await_completion() → None[source]¶

Wait for the task to complete - either through stopping naturally or being cancelled.

Returns:: None

cancel() → None[source]¶

Cancel the task. Does nothing if the task is already stopped or canceled, or has not been started.

Returns:: None

start() → None[source]¶

Start the task. Does nothing if the task is already running.

Returns:: None
Raises:: RuntimeError – When trying to start a task that has not been added to a client, or the client it has been added to has not been started yet.

stop() → None[source]¶

Stop the task. Does nothing if the task is already stopped, or has not been started.

Returns:: None

property running: bool¶: Whether the task is running. A task is considered running if it has been started, but not stopped nor cancelled.

class TaskExecutionData(invocation_count: int, last_invocation_length: float, last_invoked_at: datetime | None)[source]¶

Dataclass representing the data passed to a trigger function.

invocation_count: int¶: The number of times the task has been invoked so far.

last_invocation_length: float¶: The length of the last invocation of the task. Will be -1 if the task has not been invoked yet.

last_invoked_at: datetime | None¶: The time (UTC) of the last invocation of the task or None if the task has not been invoked.

crontrigger(tab: str) → Callable[[TaskExecutionData], float][source]¶

Generates a crontab-based task trigger. Tasks will be run dependent on the given crontab. You can use a tool such as crontab.guru to aid in creating an appropriate crontab.

Parameters:: tab – The crontab to use to schedule task execution.
Returns:: The generated trigger function.

Note

The crontab is always evaluated using UTC time.

Warning

This trigger is not available unless you have the croniter requirement installed. For convenience, you can install this using the ‘[crontrigger]’ option when installing Lightbulb.

E.g. pip install hikari-lightbulb[crontrigger]

Warning

If the task execution takes longer than the interval specified by the crontab, the next execution will take place at the next suitable time after the execution completed. For example, if you specify the task should run at every 5th minute, but the execution at the 0th minute takes 6 minutes - the task would next be executed at the 10th minute instead of the 5th minute.

Example

# Crontab '* * * * *' means 'run at every minute'
@client.task(lightbulb.crontrigger("* * * * *"))
async def print_hi() -> None:
    print("HI")

uniformtrigger(seconds: int = 0, minutes: int = 0, hours: int = 0, wait_first: bool = True) → Callable[[TaskExecutionData], float][source]¶

Generates a trigger function that returns uniform intervals from the given arguments. At least one of seconds, minutes, and hours must be specified. If multiple are specified then the intervals will be combined. I.e. if hours is 1 AND seconds is 5, then the generated interval will be 3605 seconds (1 hour 5 seconds).

Parameters:

seconds – The number of seconds for the interval.
minutes – The number of minutes for the interval.
hours – The number of hours for the interval.
wait_first – Whether the first invocation of the task should wait for the interval to elapse. Defaults to True.

Returns:

The generated trigger function.

Raises:

ValueError – When all interval arguments are 0, or any interval argument is negative.

Example

@client.task(lightbulb.uniformtrigger(minutes=1))
async def print_hi() -> None:
    print("HI")