lightbulb.prefab.concurrency

exception MaxConcurrencyReached

Exception raised when a user attempts to invoke a command, but the concurrency limit has been reached.

max_concurrency(n_invocations: int, bucket: Literal['global', 'user', 'channel', 'guild'] | Callable[[Context], Snowflake | int | Coroutine[Any, Any, Snowflake | int]]) → tuple[ExecutionHook, ExecutionHook]

Creates hooks that enforce a concurrency limit for a single command. The created hooks raise MaxConcurrencyReached when they fail. The created hooks are run during the MAX_CONCURRENCY and POST_INVOKE execution steps. As this returns multiple hooks, you should unpack them into the hooks list for your command - see the example for details.

Parameters:

• n_invocations – The number of invocations permitted to be running at the same time.

• bucket – The bucket which invocations should be limited within. Accepts the same values that the cooldowns do.

Returns:

The created hooks.

Warning

DO NOT use the same hooks for multiple commands - this will cause the concurrency limit to be shared between them. Make sure you call this function a single time for each command you wish to enforce a concurrency limit with.

Warning

If you add any other checks for the POST_INVOKE execution step, you should probably specify them before these in the hooks list for the command - otherwise the concurrency limit will be decreased before the other POST_INVOKE hooks have been run.

Example

class YourCommand(
    ...,
    hooks=[..., *lightbulb.prefab.max_concurrency(1, "global")]
):
    ...