tanjun.dependencies#

Get an entry from this cache by ID.

Parameters:

• key (_KeyT) –

  Unique key of the entry to get; this will often be a snowflake.

• default (_DefaultT, default: ... ) –

  The default value to return if an entry wasn't found.

  If provided then no errors will be raised when no entry is found.

Returns:

• _ValueT | _DefaultT –

  The found entry or the default if any was provided.

Raises:

• CacheMissError –

  If the entry wasn't found.

  This won't be raised if default is passed.

• EntryNotFound –

  If the entry wasn't found and the the entry definitely doesn't exist.

  This won't be raised if default is passed.

  This is a specialisation of CacheMissError and thus may be caught as CacheMissError and otherwise would need to be beforeCacheMissError` in a try, multiple catch statement.

Asynchronously iterate over the globally cached entries for this resource.

Returns:

• CacheIterator[_ValueT] –

  An asynchronous iterator of the entries cached globally for this resource.

Get the length of the target resource.

Returns:

• int –

  The length of the targeted resource.

Get an entry from this cache for a specific channel by ID.

Parameters:

• channel_id (Snowflakeish) –

  ID of the channel to get an entry for.

• key (_KeyT) –

  Unique key of the entry to get; this will usually be a snowflake.

• default (_DefaultT, default: ... ) –

  The default value to return if an entry wasn't found.

  If provided then no errors will be raised when no entry is found.

Returns:

• _ValueT | _DefaultT –

  The found entry or the default if any was provided.

Raises:

• CacheMissError –

  If the entry wasn't found.

  This won't be raised if default is passed.

• EntryNotFound –

  If the entry wasn't found and the the entry definitely doesn't exist.

  This won't be raised if default is passed.

  This is a specialisation of CacheMissError and thus may be caught as CacheMissError and otherwise would need to be beforeCacheMissError` in a try, multiple catch statement.

Asynchronously iterate over the globally cached entries for this resource.

Returns:

• CacheIterator[_ValueT] –

  An asynchronous iterator of the entries cached globally for this resource.

Asynchronously iterate over the entries entries cached for a channel.

Parameters:

• channel_id (Snowflakeish) –

  ID of the channel to iterate over the entries cached for.

Returns:

• CacheIterator[_ValueT] –

  An asynchronous iterator of the entries cached for the specified channel.

Get an entry from this cache for a specific guild by ID.

Parameters:

• guild_id (Snowflakeish) –

  ID of the guild to get an entry for.

• key (_KeyT) –

  Unique key of the entry to get; this will usually be a snowflake.

• default (_DefaultT, default: ... ) –

  The default value to return if an entry wasn't found.

  If provided then no errors will be raised when no entry is found.

Returns:

• _ValueT | _DefaultT –

  The found entry or the default if any was provided.

Raises:

• CacheMissError –

  If the entry wasn't found.

  This won't be raised if default is passed.

• EntryNotFound –

  If the entry wasn't found and the the entry definitely doesn't exist.

  This won't be raised if default is passed.

  This is a specialisation of CacheMissError and thus may be caught as CacheMissError and otherwise would need to be beforeCacheMissError` in a try, multiple catch statement.

Asynchronously iterate over the globally cached entries for this resource.

Returns:

• CacheIterator[_ValueT] –

  An asynchronous iterator of the entries cached globally for this resource.

Asynchronously iterate over the entries entries cached for a guild.

Parameters:

• guild_id (Snowflakeish) –

  ID of the guild to iterate over the entries cached for.

Returns:

• CacheIterator[_ValueT] –

  An asynchronous iterator of the entries cached for the specified guild.

Get the entry.

Parameters:

• default (_DefaultT, default: ... ) –

  The default value to return if an entry wasn't found.

  If provided then no errors will be raised when no entry is found.

Returns:

• _ValueT | _DefaultT –

  The found entry or the default if any was provided.

Raises:

• CacheMissError –

  If the entry wasn't found.

  This won't be raised if default is passed.

• EntryNotFound –

  If the entry wasn't found and the the entry definitely doesn't exist.

  This won't be raised if default is passed.

  This is a specialisation of CacheMissError and thus may be caught as CacheMissError and otherwise would need to be beforeCacheMissError` in a try, multiple catch statement.

Initiate a new lazy constant.

Parameters:

• callback (CallbackSig[_T]) –

  Callback used to resolve this to a constant value.

  This supports dependency injection and may either be sync or asynchronous.

Acquire this lazy constant as an asynchronous lock.

This is used to ensure that the value is only generated once and should be kept acquired until LazyConstant.set_value has been called.

Returns:

• AbstractAsyncContextManager[Any] –

  Context manager that can be used to acquire the lock.

Get the value of this constant if set, else None.

Clear the internally stored value.

Set the constant value.

Parameters:

• value (_T) –

  The value to set.

Raises:

• RuntimeError –

  If the constant has already been set.

Release a concurrency lock on a bucket.

Parameters:

• bucket_id (str) –

  The concurrency bucket to release.

• ctx (Context) –

  The context to release.

Raises:

• ResourceNotTracked –

  If the passed bucket ID and context are not currently acquired.

Try to acquire a concurrency lock on a bucket.

Parameters:

• bucket_id (str) –

  The concurrency bucket to acquire.

• ctx (Context) –

  The context to acquire this bucket lock with.

Raises:

• ResourceDepleted –

  If the concurrency lock couldn't be acquired.

Acquire an concurrency lock on a bucket through an async context manager.

Parameters:

• bucket_id (str) –

  The concurrency bucket to acquire.

• ctx (Context) –

  The context to acquire this resource lock with.

• error (Callable[[], Exception], default: lambda: CommandError('This resource is currently busy; please try again later.') ) –

  Callback which returns the error that's raised when the lock couldn't be acquired due to being at it's limit.

  This will be raised on entering the returned context manager and defaults to an English command error.

Returns:

• AbstractAsyncContextManager[None] –

  The context manager which'll acquire and release this concurrency lock.

Raises:

• CommandError –

  The default error that's raised while entering the returned async context manager if it couldn't acquire the lock.

Release a concurrency lock on a bucket.

Parameters:

• bucket_id (str) –

  The concurrency bucket to release.

• ctx (Context) –

  The context to release.

Raises:

• ResourceNotTracked –

  If the passed bucket ID and context are not currently acquired.

Try to acquire a concurrency lock on a bucket.

Parameters:

• bucket_id (str) –

  The concurrency bucket to acquire.

• ctx (Context) –

  The context to acquire this bucket lock with.

Raises:

• ResourceDepleted –

  If the concurrency lock couldn't be acquired.

Release a bucket's cooldown.

Parameters:

• bucket_id (str) –

  The bucket to decrement a cooldown for.

• ctx (Context) –

  Context of the command call.

Raises:

• ResourceNotTracked –

  If the passed bucket ID and context are not currently contributing towards the cooldown.

Increment a bucket's cooldown.

Parameters:

• bucket_id (str) –

  The bucket to increment a cooldown for.

• ctx (Context) –

  Context of the command call

Raises:

• CooldownDepleted –

  If the cooldown couldn't be acquired.

Acquire a cooldown lock on a bucket through an async context manager.

Parameters:

• bucket_id (str) –

  The cooldown bucket to acquire.

• ctx (Context) –

  The context to acquire this resource lock with.

• error (Callable[[Optional[datetime]], Exception], default: lambda cooldown: CommandError('This command is currently in cooldown.' + f' Try again {from_datetime(cooldown, style='R')}.' if cooldown else '') ) –

  Callback which returns the error that's raised when the lock couldn't be acquired due to it being on cooldown.

  This will be raised on entering the returned context manager and defaults to an English command error.

Returns:

• AbstractAsyncContextManager[None] –

  The context manager which'll acquire and release this cooldown lock.

Raises:

• CommandError –

  The default error that's raised while entering the returned async context manager if it couldn't acquire the lock.

Deprecated method.

Deprecated function for incrementing a cooldown.

Use AbstractCooldownManager.acquire and AbstractCooldownManager.release.

Release a bucket's cooldown.

Parameters:

• bucket_id (str) –

  The bucket to decrement a cooldown for.

• ctx (Context) –

  Context of the command call.

Raises:

• ResourceNotTracked –

  If the passed bucket ID and context are not currently contributing towards the cooldown.

Increment a bucket's cooldown.

Parameters:

• bucket_id (str) –

  The bucket to increment a cooldown for.

• ctx (Context) –

  Context of the command call

Raises:

• CooldownDepleted –

  If the cooldown couldn't be acquired.

Initialise a concurrency post-execution hook.

Parameters:

• bucket_id (str) –

  The concurrency limit bucket's ID.

Initialise a concurrency pre-execution hook.

Parameters:

• bucket_id (str) –

  The concurrency limit bucket's ID.

• error (Optional[Callable[[str], Exception]], default: None ) –

  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 takes priority over error_message.

• error_message (Union[str, Mapping[str, str]], default: 'This resource is currently busy; please try again later.' ) –

  The error message to send in response as a command error if this fails to acquire the concurrency limit.

  This supports localisation and uses the check name "tanjun.concurrency" for global overrides.

Initialise a pre-execution cooldown command hook.

Parameters:

• bucket_id (str) –

  The cooldown bucket's ID.

• error (Optional[Callable[[str, Optional[datetime]], Exception]], default: None ) –

  Callback used to create a custom error to raise if the check fails.

  This should two arguments one of type str and datetime.datetime | None where the first is the limiting bucket's ID and the second is when said bucket can be used again if known.

  This takes priority over error_message.

• error_message (Union[str, Mapping[str, str]], default: 'This command is currently in cooldown. Try again {cooldown}.' ) –

  The error message to send in response as a command error if the check fails.

  This supports localisation and uses the check name "tanjun.cooldown" for global overrides.

• unknown_message (Union[str, Mapping[str, str], None], default: None ) –

  Response error message for when cooldown is unknown.

  This supports localisation and uses the check name "tanjun.cooldown_unknown" for global overrides.

  This defaults to error_message but takes no format args.

• owners_exempt (bool, default: True ) –

  Whether owners should be exempt from the cooldown.

Acquire an concurrency lock on a bucket through an async context manager.

Parameters:

• bucket_id (str) –

  The concurrency bucket to acquire.

• ctx (Context) –

  The context to acquire this resource lock with.

• error (Callable[[], Exception], default: lambda: CommandError('This resource is currently busy; please try again later.') ) –

  Callback which returns the error that's raised when the lock couldn't be acquired due to being at it's limit.

  This will be raised on entering the returned context manager and defaults to an English command error.

Returns:

• AbstractAsyncContextManager[None] –

  The context manager which'll acquire and release this concurrency lock.

Raises:

• CommandError –

  The default error that's raised while entering the returned async context manager if it couldn't acquire the lock.

Add this concurrency manager to a tanjun client.

Parameters:

• client (Client) –

  The client to add this concurrency manager to.

Stop the concurrency manager.

Raises:

• RuntimeError –

  If the concurrency manager is not running.

Disable a concurrency limit bucket.

This will stop the bucket from ever hitting a concurrency limit and also prevents the bucket from defaulting.

Parameters:

• bucket_id (str) –

  The bucket to disable.

Returns:

• Self –

  This concurrency manager to allow for chaining.

Start the concurrency manager.

Raises:

• RuntimeError –

  If the concurrency manager is already running. If called in a thread with no running event loop.

Set the concurrency limit for a specific bucket.

Parameters:

• bucket_id (str) –

  The ID of the bucket to set the concurrency limit for.

• resource (BucketResource) –

  The type of resource to target for the concurrency limit.

• limit (int) –

  The maximum number of concurrent uses to allow.

Returns:

• Self –

  The concurrency manager to allow call chaining.

Raises:

• ValueError –

  If any of the following cases are met:

  • If an invalid resource is passed.
  • If limit is less 0 or negative.

Set a custom concurrency limit resource.

Parameters:

• resource (AbstractConcurrencyBucket) –

  Object which handles the concurrency limits for these buckets.

• bucket_ids (str, default: () ) –

  IDs of buckets to set this custom resource for.

Returns:

• Self –

  The concurrency manager to allow call chaining.

Examples:

class CustomBucket(tanjun.dependencies.AbstractConcurrencyBucket):
    __slots__ = ()

    async def try_acquire(
        self, bucket_id: str, ctx: tanjun.abc.Context, /
    ) -> bool:
        # ResourceDepleted should be raised if this couldn't be acquired.
        raise tanjun.dependencies.ResourceDepleted

    async def release(
        self, bucket_id: str, ctx: tanjun.abc.Context, /
    ) -> None:
        ...

(
    tanjun.dependencies.InMemoryConcurrencyLimiter()
    .set_custom_bucket(CustomBucket(), "BUCKET_ID", "OTHER_BUCKET_ID")
)

Acquire a cooldown lock on a bucket through an async context manager.

Parameters:

• bucket_id (str) –

  The cooldown bucket to acquire.

• ctx (Context) –

  The context to acquire this resource lock with.

• error (Callable[[Optional[datetime]], Exception], default: lambda cooldown: CommandError('This command is currently in cooldown.' + f' Try again {from_datetime(cooldown, style='R')}.' if cooldown else '') ) –

  Callback which returns the error that's raised when the lock couldn't be acquired due to it being on cooldown.

  This will be raised on entering the returned context manager and defaults to an English command error.

Returns:

• AbstractAsyncContextManager[None] –

  The context manager which'll acquire and release this cooldown lock.

Raises:

• CommandError –

  The default error that's raised while entering the returned async context manager if it couldn't acquire the lock.

Add this cooldown manager to a tanjun client.

Parameters:

• client (Client) –

  The client to add this cooldown manager to.

Stop the cooldown manager.

Raises:

• RuntimeError –

  If the cooldown manager is not running.

Disable a cooldown bucket.

This will stop the bucket from ever hitting a cooldown and also prevents the bucket from defaulting.

Parameters:

• bucket_id (str) –

  The bucket to disable.

Returns:

• Self –

  This cooldown manager to allow for chaining.

Deprecated function for incrementing a cooldown.

Use AbstractCooldownManager.acquire and AbstractCooldownManager.release.

Start the cooldown manager.

Raises:

• RuntimeError –

  If the cooldown manager is already running. If called in a thread with no running event loop.

Set the cooldown for a specific bucket.

Parameters:

• bucket_id (str) –

  The ID of the bucket to set the cooldown for.

• resource (BucketResource) –

  The type of resource to target for the cooldown.

• limit (int) –

  The number of uses per cooldown period.

• reset_after (Union[int, float, timedelta]) –

  The cooldown period.

Returns:

• Self –

  The cooldown manager to allow call chaining.

Raises:

• ValueError –

  If any of the following cases are met:

  • If an invalid resource is passed.
  • If reset_after or limit are negative, 0 or invalid.
  • If limit is less 0 or negative.

Set a custom cooldown limit resource.

Parameters:

• resource (AbstractCooldownBucket) –

  Object which handles the cooldowns for these buckets.

• bucket_ids (str, default: () ) –

  IDs of buckets to set this custom resource for.

Returns:

• Self –

  The cooldown manager to allow call chaining.

Examples:

class CustomBucket(tanjun.dependencies.AbstractCooldownBucket):
    __slots__ = ()

    async def try_acquire(
        self, bucket_id: str, ctx: tanjun.abc.Context, /
    ) -> None:
        # CooldownDepleted should be raised if this couldn't be acquired.
        raise tanjun.dependencies.CooldownDepleted(None)

    async def release(
        self, bucket_id: str, ctx: tanjun.abc.Context, /
    ) -> None:
        ...

(
    tanjun.dependencies.InMemoryCooldownManager()
    .set_custom_bucket(CustomBucket(), "BUCKET_ID", "OTHER_BUCKET_ID")
)

Get all the localisation variants for an identifier.

Localise a string with the given identifier and arguments.

Parameters:

• identifier (str) –

  The unique identifier of the string to localise.

  This may be in any format but the formats used by the standard implementations can be found at client-localiser.

• tag (str) –

  The "IETF lang tag" to localise the string to.

  This should usually be a hikari.Locale.

• **kwargs (Any, default: {} ) –

  Key-word arguments to pass to the string as format args.

Returns:

• str –

  The localised string.

Alias for AbstractLocaliser.localise.

Initialise a new BasicLocaliser.

Add this global localiser to a tanjun client.

Parameters:

• client (Client) –

  The client to add this global localiser to.

Alias for AbstractLocaliser.localise.

Set the variants for a localised field.

Parameters:

• identifier (str) –

  Identifier of the field to set the localised variants for.

  This may be in any format but the formats used by the standard implementations can be found at client-localiser.

• variants (Optional[Mapping[str, str]], default: None ) –

  Mapping of hikari.Locales to the localised values.

Returns:

• Self –

  The localiser object to enable chained calls.

Check whether this object is owned by the given object.

Parameters:

• client (Client) –

  The Tanjun client this check is being called by.

• user (User) –

  The user to check ownership for.

Returns:

• bool –

  Whether the bot is owned by the provided user.

Initiate a new owner check dependency.

Parameters:

• expire_after (Union[timedelta, int, float], default: timedelta(minutes=5) ) –

  The amount of time to cache application owner data for in seconds.

  This is only applicable if rest is also passed.

• fallback_to_application (bool, default: True ) –

  Whether this check should fallback to checking the application's owners if the user isn't in owners.

  This only works when the bot's rest client is bound to a Bot token or if tanjun.dependencies.SingleStoreCache[hikari.Application] is available.

• owners (Optional[SnowflakeishSequence[User]], default: None ) –

  Sequence of objects and IDs of the users that are allowed to use the bot's owners-only commands.

Initialise a hot reloader.

Parameters:

• commands_guild (Optional[SnowflakeishOr[PartialGuild]], default: None ) –

  Object or ID of the guild to declare commands in if redeclare_cmds_after is not None.

• interval (Union[int, float, timedelta], default: timedelta(microseconds=500000) ) –

  How often this should scan files and directories for changes in seconds.

• redeclare_cmds_after (Union[int, float, timedelta, None], default: timedelta(seconds=10) ) –

  How often to redeclare application commands after a change to the commands is detected.

  If None is passed here then this will not redeclare the application's commands.

• unload_on_delete (bool, default: True ) –

  Whether this should unload modules when their relevant file is deleted.

Add a directory for this hot reloader to track.

Parameters:

• directory (Union[str, Path]) –

  Path of the directory to hot reload.

• namespace (Optional[str], default: None ) –

  The python namespace this directory's modules should be imported from, if applicable.

  This work as {namespace}.{file.name.removesuffix(".py")} and will have the same behaviour as when a str is passed to Client.load_modules if passed.

  If left as None then this will have the same behaviour as when a pathlib.Path is passed to Client.load_modules.

Returns:

• Self –

  The hot reloader to enable chained calls.

Raises:

• FileNotFoundError –

  If the directory cannot be found

Asynchronous variant of HotReloader.add_directory.

Unlike HotReloader.add_directory, this method will run blocking code in a background thread.

For more information on the behaviour of this method see the documentation for HotReloader.add_directory.

Add modules for this hot reloader to track.

Parameters:

• *paths (Union[str, Path], default: () ) –

  Module paths for this hot reloader to track.

  This has the same behaviour as [tanjun.abc.Client.load_modules][ for how [pathlib.Path][] and str are treated.

Raises:

• FileNotFoundError –

  If the module's file doesn't exist anymore.

• ModuleNotFoundError –

  If the str module path cannot be imported.

Asynchronous variant of HotReloader.add_modules.

Unlike HotReloader.add_modules, this method will run blocking code in a background thread.

For more information on the behaviour of this method see the documentation for HotReloader.add_modules.

Add this to a tanjun.abc.Client instance.

This registers start and closing callbacks which handle the lifetime of this and adds this as a type dependency.

Parameters:

• client (Client) –

  The client to link this hot reloader to.

Manually scan this hot reloader's tracked modules for changes.

Parameters:

• client (Client) –

  The client to reload and unload modules in.

Start the hot reloader.

Raises:

• RuntimeError –

  If the hot reloader is already running.

Stop the hot reloader.

Raises:

• RuntimeError –

  If the hot reloader isn't running.