From ac6a8c039ec45e846c63bf4e4edba9fff3d9bbd6 Mon Sep 17 00:00:00 2001 From: Faster Speeding Date: Tue, 4 Oct 2022 20:52:41 +0100 Subject: [PATCH] Add support for sync DI to sync callbacks --- tanjun/checks.py | 103 ++++++++++++++++++++------------ tanjun/clients.py | 75 +++++++++++------------ tanjun/dependencies/limiters.py | 29 ++++++--- 3 files changed, 120 insertions(+), 87 deletions(-) diff --git a/tanjun/checks.py b/tanjun/checks.py index 356acf049..643feff40 100644 --- a/tanjun/checks.py +++ b/tanjun/checks.py @@ -68,6 +68,15 @@ from . import permissions from ._internal import localisation +if typing.TYPE_CHECKING: + import typing_extensions + + _P = typing_extensions.ParamSpec("_P") + + _PermissionErrorSigBase = collections.Callable[typing_extensions.Concatenate[hikari.Permissions, _P], Exception] + _PermissionErrorSig = _PermissionErrorSigBase[...] + + _CommandT = typing.TypeVar("_CommandT", bound="tanjun.ExecutableCommand[typing.Any]") # This errors on earlier 3.9 releases when not quotes cause dumb handling of the [_CommandT] list _CallbackReturnT = typing.Union[_CommandT, "collections.Callable[[_CommandT], _CommandT]"] @@ -116,7 +125,7 @@ def _handle_result( ) -> bool: if not result: if self._error: - raise self._error(*args) from None + raise ctx.call_with_di(self._error, args) from None if self._halt_execution: raise errors.HaltExecution from None if self._error_message: @@ -137,7 +146,7 @@ class OwnerCheck(_Check): def __init__( self, *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None] = "Only bot owners can use this command", halt_execution: bool = False, ) -> None: @@ -148,7 +157,8 @@ def __init__( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positional arguments, supports sync DI and takes + priority over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -216,7 +226,7 @@ class NsfwCheck(_Check): def __init__( self, *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[ str, collections.Mapping[str, str], None ] = "Command can only be used in NSFW channels", @@ -229,7 +239,8 @@ def __init__( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positional arguments, supports sync DI and takes + priority over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -269,7 +280,7 @@ class SfwCheck(_Check): def __init__( self, *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[ str, collections.Mapping[str, str], None ] = "Command can only be used in SFW channels", @@ -282,7 +293,8 @@ def __init__( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positonal arguments, supports sync DI and takes + priority over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -322,7 +334,7 @@ class DmCheck(_Check): def __init__( self, *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None] = "Command can only be used in DMs", halt_execution: bool = False, ) -> None: @@ -333,7 +345,8 @@ def __init__( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positonal arguments, supports sync DI and takes + priority over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -369,7 +382,7 @@ class GuildCheck(_Check): def __init__( self, *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[ str, collections.Mapping[str, str], None ] = "Command can only be used in guild channels", @@ -382,7 +395,8 @@ def __init__( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positonal arguments, supports sync DI and takes + priority over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -420,7 +434,7 @@ def __init__( permissions: typing.Union[hikari.Permissions, int], /, *, - error: typing.Optional[collections.Callable[[hikari.Permissions], Exception]] = None, + error: typing.Optional[_PermissionErrorSig] = None, error_message: typing.Union[ str, collections.Mapping[str, str], None ] = "You don't have the permissions required to use this command", @@ -436,7 +450,8 @@ def __init__( Callback used to create a custom error to raise if the check fails. This should take 1 positional argument of type [hikari.permissions.Permissions][] - which represents the missing permissions required for this command to run. + which represents the missing permissions required for this command + to run, return an [Exception][] to raise, and supports sync DI. This takes priority over `error_message`. error_message @@ -499,7 +514,7 @@ def __init__( permissions: typing.Union[hikari.Permissions, int], /, *, - error: typing.Optional[collections.Callable[[hikari.Permissions], Exception]] = None, + error: typing.Optional[_PermissionErrorSig] = None, error_message: typing.Union[ str, collections.Mapping[str, str], None ] = "Bot doesn't have the permissions required to run this command", @@ -515,7 +530,8 @@ def __init__( Callback used to create a custom error to raise if the check fails. This should take 1 positional argument of type [hikari.permissions.Permissions][] - which represents the missing permissions required for this command to run. + which represents the missing permissions required for this command + to run, return an [Exception][] to raise, and supports sync DI. This takes priority over `error_message`. error_message @@ -570,7 +586,7 @@ def with_dm_check(command: _CommandT, /) -> _CommandT: @typing.overload def with_dm_check( *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None] = "Command can only be used in DMs", follow_wrapped: bool = False, halt_execution: bool = False, @@ -582,7 +598,7 @@ def with_dm_check( command: typing.Optional[_CommandT] = None, /, *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None] = "Command can only be used in DMs", follow_wrapped: bool = False, halt_execution: bool = False, @@ -596,7 +612,8 @@ def with_dm_check( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positonal arguments, supports sync DI and takes priority + over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -631,7 +648,7 @@ def with_guild_check(command: _CommandT, /) -> _CommandT: @typing.overload def with_guild_check( *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[ str, collections.Mapping[str, str], None ] = "Command can only be used in guild channels", @@ -645,7 +662,7 @@ def with_guild_check( command: typing.Optional[_CommandT] = None, /, *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[ str, collections.Mapping[str, str], None ] = "Command can only be used in guild channels", @@ -661,7 +678,8 @@ def with_guild_check( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positonal arguments, supports sync DI and takes priority + over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -696,7 +714,7 @@ def with_nsfw_check(command: _CommandT, /) -> _CommandT: @typing.overload def with_nsfw_check( *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None] = "Command can only be used in NSFW channels", follow_wrapped: bool = False, halt_execution: bool = False, @@ -708,7 +726,7 @@ def with_nsfw_check( command: typing.Optional[_CommandT] = None, /, *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None] = "Command can only be used in NSFW channels", follow_wrapped: bool = False, halt_execution: bool = False, @@ -722,7 +740,8 @@ def with_nsfw_check( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positonal arguments, supports sync DI and takes priority + over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -757,7 +776,7 @@ def with_sfw_check(command: _CommandT, /) -> _CommandT: @typing.overload def with_sfw_check( *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None] = "Command can only be used in SFW channels", follow_wrapped: bool = False, halt_execution: bool = False, @@ -769,7 +788,7 @@ def with_sfw_check( command: typing.Optional[_CommandT] = None, /, *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None] = "Command can only be used in SFW channels", follow_wrapped: bool = False, halt_execution: bool = False, @@ -783,7 +802,8 @@ def with_sfw_check( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positonal arguments, supports sync DI and takes priority + over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -818,7 +838,7 @@ def with_owner_check(command: _CommandT, /) -> _CommandT: @typing.overload def with_owner_check( *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None] = "Only bot owners can use this command", follow_wrapped: bool = False, halt_execution: bool = False, @@ -830,7 +850,7 @@ def with_owner_check( command: typing.Optional[_CommandT] = None, /, *, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None] = "Only bot owners can use this command", follow_wrapped: bool = False, halt_execution: bool = False, @@ -844,7 +864,8 @@ def with_owner_check( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positonal arguments, supports sync DI and takes priority + over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -874,7 +895,7 @@ def with_owner_check( def with_author_permission_check( permissions: typing.Union[hikari.Permissions, int], *, - error: typing.Optional[collections.Callable[[hikari.Permissions], Exception]] = None, + error: typing.Optional[_PermissionErrorSig] = None, error_message: typing.Union[ str, collections.Mapping[str, str], None ] = "You don't have the permissions required to use this command", @@ -895,7 +916,8 @@ def with_author_permission_check( Callback used to create a custom error to raise if the check fails. This should take 1 positional argument of type [hikari.permissions.Permissions][] - which represents the missing permissions required for this command to run. + which represents the missing permissions required for this command to + run, return an [Exception][] to raise, and supports sync DI. This takes priority over `error_message`. error_message @@ -929,7 +951,7 @@ def with_author_permission_check( def with_own_permission_check( permissions: typing.Union[hikari.Permissions, int], *, - error: typing.Optional[collections.Callable[[hikari.Permissions], Exception]] = None, + error: typing.Optional[_PermissionErrorSig] = None, error_message: typing.Union[ str, collections.Mapping[str, str], None ] = "Bot doesn't have the permissions required to run this command", @@ -950,7 +972,8 @@ def with_own_permission_check( Callback used to create a custom error to raise if the check fails. This should take 1 positional argument of type [hikari.permissions.Permissions][] - which represents the missing permissions required for this command to run. + which represents the missing permissions required for this command to + run, return an [Exception][] to raise, and supports sync DI. This takes priority over `error_message`. error_message @@ -1076,7 +1099,7 @@ class _AnyChecks(_Check): def __init__( self, checks: list[tanjun.CheckSig], - error: typing.Optional[collections.Callable[[], Exception]], + error: typing.Optional[collections.Callable[..., Exception]], error_message: typing.Union[str, collections.Mapping[str, str], None], halt_execution: bool, suppress: tuple[type[Exception], ...], @@ -1109,7 +1132,7 @@ def any_checks( check: tanjun.CheckSig, /, *checks: tanjun.CheckSig, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None], halt_execution: bool = False, suppress: tuple[type[Exception], ...] = (errors.CommandError, errors.HaltExecution), @@ -1128,7 +1151,8 @@ def any_checks( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positonal arguments, supports sync DI and takes priority + over `error_message`. error_message The error message to send in response as a command error if the check fails. @@ -1153,7 +1177,7 @@ def with_any_checks( check: tanjun.CheckSig, /, *checks: tanjun.CheckSig, - error: typing.Optional[collections.Callable[[], Exception]] = None, + error: typing.Optional[collections.Callable[..., Exception]] = None, error_message: typing.Union[str, collections.Mapping[str, str], None], follow_wrapped: bool = False, halt_execution: bool = False, @@ -1173,7 +1197,8 @@ def with_any_checks( error Callback used to create a custom error to raise if the check fails. - This takes priority over `error_message`. + This takes no positonal arguments, supports sync DI and takes priority + over `error_message`. error_message The error message to send in response as a command error if the check fails. diff --git a/tanjun/clients.py b/tanjun/clients.py index b712655d8..a392b6aef 100644 --- a/tanjun/clients.py +++ b/tanjun/clients.py @@ -143,6 +143,11 @@ def __call__( ) -> context.SlashContext: raise NotImplementedError + _LoaderSigBase = collections.Callable[typing_extensions.Concatenate[_T, _P], None] + _LoaderSig = _LoaderSigBase[_T, ...] + _LoaderSigT = typing.TypeVar("_LoaderSigT", bound=_LoaderSig[tanjun.Client]) + _StdLoaderSigT = typing.TypeVar("_StdLoaderSigT", bound=_LoaderSig["Client"]) + PrefixGetterSig = collections.Callable[..., collections.Coroutine[typing.Any, typing.Any, collections.Iterable[str]]] """Type hint of a callable used to get the prefix(es) for a specific guild. @@ -184,10 +189,12 @@ def load(self, client: tanjun.Client, /) -> bool: if not isinstance(client, Client): raise ValueError("This loader requires instances of the standard Client implementation") - self._callback(client) + client.injector.call_with_di(self._callback, client) else: - typing.cast("collections.Callable[[tanjun.Client], None]", self._callback)(client) + client.injector.call_with_di( + typing.cast("collections.Callable[[tanjun.Client], None]", self._callback), client + ) return True @@ -224,54 +231,46 @@ def unload(self, client: tanjun.Client, /) -> bool: if not isinstance(client, Client): raise ValueError("This unloader requires instances of the standard Client implementation") - self._callback(client) + client.injector.call_with_di(self._callback, client) else: - typing.cast("collections.Callable[[tanjun.Client], None]", self._callback)(client) + client.injector.call_with_di( + typing.cast("collections.Callable[[tanjun.Client], None]", self._callback), client + ) return True @typing.overload -def as_loader( - callback: collections.Callable[[Client], None], /, *, standard_impl: typing.Literal[True] = True -) -> collections.Callable[[Client], None]: +def as_loader(callback: _StdLoaderSigT, /, *, standard_impl: typing.Literal[True] = True) -> _StdLoaderSigT: ... @typing.overload -def as_loader( - *, standard_impl: typing.Literal[True] = True -) -> collections.Callable[[collections.Callable[[Client], None]], collections.Callable[[Client], None]]: +def as_loader(*, standard_impl: typing.Literal[True] = True) -> collections.Callable[[_StdLoaderSigT], _StdLoaderSigT]: ... @typing.overload -def as_loader( - callback: collections.Callable[[tanjun.Client], None], /, *, standard_impl: typing.Literal[False] -) -> collections.Callable[[tanjun.Client], None]: +def as_loader(callback: _LoaderSigT, /, *, standard_impl: typing.Literal[False]) -> _LoaderSigT: ... @typing.overload -def as_loader( - *, standard_impl: typing.Literal[False] -) -> collections.Callable[[collections.Callable[[tanjun.Client], None]], collections.Callable[[tanjun.Client], None]]: +def as_loader(*, standard_impl: typing.Literal[False]) -> collections.Callable[[_LoaderSigT], _LoaderSigT]: ... def as_loader( - callback: typing.Union[ - collections.Callable[[tanjun.Client], None], collections.Callable[[Client], None], None - ] = None, + callback: typing.Union[_LoaderSigT, _StdLoaderSigT, None] = None, /, *, standard_impl: bool = True, ) -> typing.Union[ - collections.Callable[[tanjun.Client], None], - collections.Callable[[Client], None], - collections.Callable[[collections.Callable[[Client], None]], collections.Callable[[Client], None]], - collections.Callable[[collections.Callable[[tanjun.Client], None]], collections.Callable[[tanjun.Client], None]], + _LoaderSigT, + _StdLoaderSigT, + collections.Callable[[_StdLoaderSigT], _StdLoaderSigT], + collections.Callable[[_LoaderSigT], _LoaderSigT], ]: """Mark a callback as being used to load Tanjun components from a module. @@ -287,6 +286,8 @@ def as_loader( [tanjun.abc.Client][] if `standard_impl` is [False][]), return nothing and will be expected to initiate and add utilities such as components to the provided client. + + This supports sync DI. standard_impl Whether this loader should only allow instances of [tanjun.Client][] as opposed to [tanjun.abc.Client][]. @@ -306,45 +307,37 @@ def decorator(callback: collections.Callable[[tanjun.Client], None]) -> collecti @typing.overload -def as_unloader( - callback: collections.Callable[[Client], None], /, *, standard_impl: typing.Literal[True] = True -) -> collections.Callable[[Client], None]: +def as_unloader(callback: _StdLoaderSigT, /, *, standard_impl: typing.Literal[True] = True) -> _StdLoaderSigT: ... @typing.overload def as_unloader( *, standard_impl: typing.Literal[True] = True -) -> collections.Callable[[collections.Callable[[Client], None]], collections.Callable[[Client], None]]: +) -> collections.Callable[[_StdLoaderSigT], _StdLoaderSigT]: ... @typing.overload -def as_unloader( - callback: collections.Callable[[tanjun.Client], None], /, *, standard_impl: typing.Literal[False] -) -> collections.Callable[[tanjun.Client], None]: +def as_unloader(callback: _LoaderSigT, /, *, standard_impl: typing.Literal[False]) -> _LoaderSigT: ... @typing.overload -def as_unloader( - *, standard_impl: typing.Literal[False] -) -> collections.Callable[[collections.Callable[[tanjun.Client], None]], collections.Callable[[tanjun.Client], None]]: +def as_unloader(*, standard_impl: typing.Literal[False]) -> collections.Callable[[_LoaderSigT], _LoaderSigT]: ... def as_unloader( - callback: typing.Union[ - collections.Callable[[Client], None], collections.Callable[[tanjun.Client], None], None - ] = None, + callback: typing.Union[_StdLoaderSigT, _LoaderSigT, None] = None, /, *, standard_impl: bool = True, ) -> typing.Union[ - collections.Callable[[Client], None], - collections.Callable[[tanjun.Client], None], - collections.Callable[[collections.Callable[[Client], None]], collections.Callable[[Client], None]], - collections.Callable[[collections.Callable[[tanjun.Client], None]], collections.Callable[[tanjun.Client], None]], + _StdLoaderSigT, + _LoaderSigT, + collections.Callable[[_StdLoaderSigT], _StdLoaderSigT], + collections.Callable[[_LoaderSigT], _LoaderSigT], ]: """Mark a callback as being used to unload a module's utilities from a client. @@ -362,6 +355,8 @@ def as_unloader( [tanjun.abc.Client][] if `standard_impl` is [False][]), return nothing and will be expected to remove utilities such as components from the provided client. + + This supports sync DI. standard_impl Whether this unloader should only allow instances of [tanjun.Client][] as opposed to [tanjun.abc.Client][]. diff --git a/tanjun/dependencies/limiters.py b/tanjun/dependencies/limiters.py index 49d620124..a24c765f4 100644 --- a/tanjun/dependencies/limiters.py +++ b/tanjun/dependencies/limiters.py @@ -67,10 +67,19 @@ from . import owners if typing.TYPE_CHECKING: + import typing_extensions + _CommandT = typing.TypeVar("_CommandT", bound="tanjun.ExecutableCommand[typing.Any]") _OtherCommandT = typing.TypeVar("_OtherCommandT", bound="tanjun.ExecutableCommand[typing.Any]") _InMemoryCooldownManagerT = typing.TypeVar("_InMemoryCooldownManagerT", bound="InMemoryCooldownManager") _InMemoryConcurrencyLimiterT = typing.TypeVar("_InMemoryConcurrencyLimiterT", bound="InMemoryConcurrencyLimiter") + _P = typing_extensions.ParamSpec("_P") + + _ConcurrencyErrorSigBase = collections.Callable[typing_extensions.Concatenate[str, _P], Exception] + _ConcurrencyErrorSig = _ConcurrencyErrorSigBase[...] + _CooldownErrorSigBase = collections.Callable[typing_extensions.Concatenate[str, datetime.datetime, _P], Exception] + _CooldownErrorSig = _CooldownErrorSigBase[...] + _LOGGER: typing.Final[logging.Logger] = logging.getLogger("hikari.tanjun") @@ -665,7 +674,7 @@ def __init__( bucket_id: str, /, *, # TODO: also take ctx - error: typing.Optional[collections.Callable[[str, datetime.datetime], Exception]] = None, + error: typing.Optional[_CooldownErrorSig] = None, error_message: typing.Union[ str, collections.Mapping[str, str] ] = "This command is currently in cooldown. Try again {cooldown}.", @@ -682,7 +691,8 @@ def __init__( This should two arguments one of type [str][] and [datetime.datetime][] where the first is the limiting bucket's ID and the second is when said - bucket can be used again. + bucket can be used again, return the [Exception][] to be raised and + supports sync DI. This takes priority over `error_message`. error_message @@ -726,7 +736,7 @@ def with_cooldown( bucket_id: str, /, *, # TODO: also take ctx - error: typing.Optional[collections.Callable[[str, datetime.datetime], Exception]] = None, + error: typing.Optional[_CooldownErrorSig] = None, error_message: typing.Union[ str, collections.Mapping[str, str] ] = "This command is currently in cooldown. Try again {cooldown}.", @@ -750,7 +760,8 @@ def with_cooldown( This should two arguments one of type [str][] and [datetime.datetime][] where the first is the limiting bucket's ID and the second is when said - bucket can be used again. + bucket can be used again, return the [Exception][] to be raised, and + supports sync DI. This takes priority over `error_message`. error_message @@ -1014,7 +1025,7 @@ def __init__( bucket_id: str, /, *, - error: typing.Optional[collections.Callable[[str], Exception]] = None, # TODO: also take ctx + error: typing.Optional[_ConcurrencyErrorSig] = None, # TODO: also take ctx error_message: typing.Union[ str, collections.Mapping[str, str] ] = "This resource is currently busy; please try again later.", @@ -1028,7 +1039,8 @@ def __init__( error Callback used to create a custom error to raise if the check fails. - This should two one [str][] argument which is the limiting bucket's ID. + This should one [str][] argument which is the limiting bucket's ID, + return the [Exception][] to be raised, and supports DI. This takes priority over `error_message`. error_message @@ -1089,7 +1101,7 @@ def with_concurrency_limit( bucket_id: str, /, *, - error: typing.Optional[collections.Callable[[str], Exception]] = None, # TODO: also take ctx + error: typing.Optional[_CooldownErrorSig] = None, # TODO: also take ctx error_message: typing.Union[ str, collections.Mapping[str, str] ] = "This resource is currently busy; please try again later.", @@ -1110,7 +1122,8 @@ def with_concurrency_limit( error Callback used to create a custom error to raise if the check fails. - This should two one [str][] argument which is the limiting bucket's ID. + This should one [str][] argument which is the limiting bucket's ID, + return the [Exception][] to be raised, and supports DI. This takes priority over `error_message`. error_message