diff --git a/lightbulb/client.py b/lightbulb/client.py index 6ca06039..869c76ad 100644 --- a/lightbulb/client.py +++ b/lightbulb/client.py @@ -39,18 +39,20 @@ if t.TYPE_CHECKING: from lightbulb.commands import options + from lightbulb.internal.types import MaybeAwaitable T = t.TypeVar("T") -CommandOrGroupT = t.TypeVar("CommandOrGroupT", bound=t.Union[groups.Group, t.Type[commands.CommandBase]]) +CommandOrGroup: t.TypeAlias = t.Union[groups.Group, type[commands.CommandBase]] +CommandOrGroupT = t.TypeVar("CommandOrGroupT", bound=CommandOrGroup) CommandMapT = t.MutableMapping[hikari.Snowflakeish, t.MutableMapping[str, utils.CommandCollection]] OptionT = t.TypeVar("OptionT", bound=hikari.CommandInteractionOption) +LOGGER = logging.getLogger("lightbulb.client") DEFAULT_EXECUTION_STEP_ORDER = ( execution.ExecutionSteps.MAX_CONCURRENCY, execution.ExecutionSteps.CHECKS, execution.ExecutionSteps.COOLDOWNS, ) -LOGGER = logging.getLogger("lightbulb.client") @t.runtime_checkable @@ -82,6 +84,7 @@ class Client: to support localizations. delete_unknown_commands (:obj:`bool`): Whether to delete existing commands that the client does not have an implementation for during command syncing. + deferred_registration_callback (:obj:`~typing.Optional` [ ) """ # noqa: E501 __slots__ = ( @@ -91,8 +94,10 @@ class Client: "default_locale", "localization_provider", "delete_unknown_commands", + "deferred_registration_callback", "_di", "_localization", + "_deferred_commands", "_commands", "_application", ) @@ -105,6 +110,10 @@ def __init__( default_locale: hikari.Locale, localization_provider: localization.LocalizationProviderT, delete_unknown_commands: bool, + deferred_registration_callback: t.Callable[ + [CommandOrGroup], MaybeAwaitable[t.Union[hikari.Snowflakeish, t.Sequence[hikari.Snowflakeish]]] + ] + | None, ) -> None: super().__init__() @@ -114,9 +123,11 @@ def __init__( self.default_locale = default_locale self.localization_provider = localization_provider self.delete_unknown_commands = delete_unknown_commands + self.deferred_registration_callback = deferred_registration_callback self._di = di_.DependencyInjectionManager() + self._deferred_commands: list[CommandOrGroup] = [] self._commands: CommandMapT = collections.defaultdict(lambda: collections.defaultdict(utils.CommandCollection)) self._application: t.Optional[hikari.PartialApplication] = None @@ -126,20 +137,20 @@ def di(self) -> di_.DependencyInjectionManager: @t.overload def register( - self, *, guilds: t.Optional[t.Sequence[hikari.Snowflakeish]] = None + self, *, guilds: t.Sequence[hikari.Snowflakeish] | None = None ) -> t.Callable[[CommandOrGroupT], CommandOrGroupT]: ... @t.overload def register( - self, command: CommandOrGroupT, *, guilds: t.Optional[t.Sequence[hikari.Snowflakeish]] = None + self, command: CommandOrGroupT, *, guilds: t.Sequence[hikari.Snowflakeish] | None = None ) -> CommandOrGroupT: ... def register( self, - command: t.Optional[CommandOrGroupT] = None, + command: CommandOrGroupT | None = None, *, - guilds: t.Optional[t.Sequence[hikari.Snowflakeish]] = None, - ) -> t.Union[CommandOrGroupT, t.Callable[[CommandOrGroupT], CommandOrGroupT]]: + guilds: t.Sequence[hikari.Snowflakeish] | None = None, + ) -> CommandOrGroupT | t.Callable[[CommandOrGroupT], CommandOrGroupT]: """ Register a command or group with this client instance. Optionally, a sequence of guild ids can be provided to make the commands created in specific guilds only - overriding the value for @@ -204,6 +215,10 @@ def _inner(command_: CommandOrGroupT) -> CommandOrGroupT: return _inner + def register_deferred(self, command: CommandOrGroupT) -> CommandOrGroupT: + self._deferred_commands.append(command) + return command + async def _ensure_application(self) -> hikari.PartialApplication: if self._application is not None: return self._application @@ -223,7 +238,7 @@ async def sync_application_commands(self) -> None: @staticmethod def _get_subcommand( options: t.Sequence[OptionT], - ) -> t.Optional[OptionT]: + ) -> OptionT | None: subcommand = filter( lambda o: o.type in (hikari.OptionType.SUB_COMMAND, hikari.OptionType.SUB_COMMAND_GROUP), options ) @@ -232,24 +247,25 @@ def _get_subcommand( @t.overload def _resolve_options_and_command( self, interaction: hikari.AutocompleteInteraction - ) -> t.Optional[t.Tuple[t.Sequence[hikari.AutocompleteInteractionOption], t.Type[commands.CommandBase]]]: ... + ) -> tuple[t.Sequence[hikari.AutocompleteInteractionOption], type[commands.CommandBase]] | None: ... @t.overload def _resolve_options_and_command( self, interaction: hikari.CommandInteraction - ) -> t.Optional[t.Tuple[t.Sequence[hikari.CommandInteractionOption], t.Type[commands.CommandBase]]]: ... + ) -> tuple[t.Sequence[hikari.CommandInteractionOption], type[commands.CommandBase]] | None: ... def _resolve_options_and_command( - self, interaction: t.Union[hikari.AutocompleteInteraction, hikari.CommandInteraction] - ) -> t.Optional[ - t.Tuple[ - t.Union[t.Sequence[hikari.AutocompleteInteractionOption], t.Sequence[hikari.CommandInteractionOption]], - t.Type[commands.CommandBase], + self, interaction: hikari.AutocompleteInteraction | hikari.CommandInteraction + ) -> ( + tuple[ + t.Sequence[hikari.AutocompleteInteractionOption] | t.Sequence[hikari.CommandInteractionOption], + type[commands.CommandBase], ] - ]: + | None + ): command_path = [interaction.command_name] - subcommand: t.Union[hikari.CommandInteractionOption, hikari.AutocompleteInteractionOption, None] + subcommand: hikari.CommandInteractionOption | hikari.AutocompleteInteractionOption | None options = interaction.options or [] # TODO - check if this is hikari bug with interaction server while (subcommand := self._get_subcommand(options or [])) is not None: command_path.append(subcommand.name) @@ -287,7 +303,7 @@ def build_autocomplete_context( self, interaction: hikari.AutocompleteInteraction, options: t.Sequence[hikari.AutocompleteInteractionOption], - command_cls: t.Type[commands.CommandBase], + command_cls: type[commands.CommandBase], ) -> context_.AutocompleteContext: return context_.AutocompleteContext(self, interaction, options, command_cls) @@ -331,7 +347,7 @@ def build_command_context( self, interaction: hikari.CommandInteraction, options: t.Sequence[hikari.CommandInteractionOption], - command_cls: t.Type[commands.CommandBase], + command_cls: type[commands.CommandBase], ) -> context_.Context: """ Build a context object from the given parameters. @@ -454,7 +470,7 @@ def build_rest_autocomplete_context( self, interaction: hikari.AutocompleteInteraction, options: t.Sequence[hikari.AutocompleteInteractionOption], - command_cls: t.Type[commands.CommandBase], + command_cls: type[commands.CommandBase], response_callback: t.Callable[[hikari.api.InteractionResponseBuilder], None], ) -> context_.AutocompleteContext: return context_.RestAutocompleteContext(self, interaction, options, command_cls, response_callback) @@ -509,7 +525,7 @@ def build_rest_command_context( self, interaction: hikari.CommandInteraction, options: t.Sequence[hikari.CommandInteractionOption], - command_cls: t.Type[commands.CommandBase], + command_cls: type[commands.CommandBase], response_callback: t.Callable[[hikari.api.InteractionResponseBuilder], None], ) -> context_.Context: return context_.RestContext(self, interaction, options, command_cls(), response_callback) @@ -517,11 +533,9 @@ def build_rest_command_context( async def handle_rest_application_command_interaction( self, interaction: hikari.CommandInteraction ) -> t.AsyncGenerator[ - t.Union[ - hikari.api.InteractionDeferredBuilder, - hikari.api.InteractionMessageBuilder, - hikari.api.InteractionModalBuilder, - ], + hikari.api.InteractionDeferredBuilder + | hikari.api.InteractionMessageBuilder + | hikari.api.InteractionModalBuilder, t.Any, ]: out = self._resolve_options_and_command(interaction) @@ -560,7 +574,7 @@ def set_response(response: hikari.api.InteractionResponseBuilder) -> None: def client_from_app( - app: t.Union[GatewayClientAppT, RestClientAppT], + app: GatewayClientAppT | RestClientAppT, default_enabled_guilds: t.Sequence[hikari.Snowflakeish] = (constants.GLOBAL_COMMAND_KEY,), execution_step_order: t.Sequence[execution.ExecutionStep] = DEFAULT_EXECUTION_STEP_ORDER, default_locale: hikari.Locale = hikari.Locale.EN_US, diff --git a/lightbulb/commands/commands.py b/lightbulb/commands/commands.py index 3bce0c58..861fdb7d 100644 --- a/lightbulb/commands/commands.py +++ b/lightbulb/commands/commands.py @@ -82,7 +82,7 @@ class CommandData: invoke_method: str """The attribute name of the invoke method for the command.""" - parent: t.Optional[t.Union[groups.Group, groups.SubGroup]] = dataclasses.field(init=False, default=None) + parent: groups.Group | groups.SubGroup | None = dataclasses.field(init=False, default=None) """The group that the command belongs to, or :obj:`None` if not applicable.""" def __post_init__(self) -> None: @@ -201,13 +201,13 @@ class CommandMeta(type): run before the command invocation function is executed. Defaults to an empty set. """ - __command_types: t.ClassVar[t.Dict[type, hikari.CommandType]] = {} + __command_types: t.ClassVar[dict[type, hikari.CommandType]] = {} @staticmethod def _is_option(item: t.Any) -> bool: return isinstance(item, options_.Option) - def __new__(cls, cls_name: str, bases: t.Tuple[type, ...], attrs: t.Dict[str, t.Any], **kwargs: t.Any) -> type: + def __new__(cls, cls_name: str, bases: tuple[type, ...], attrs: dict[str, t.Any], **kwargs: t.Any) -> type: cmd_type: hikari.CommandType # Bodge because I cannot figure out how to avoid initialising all the kwargs in our # own convenience classes any other way @@ -245,7 +245,7 @@ def __new__(cls, cls_name: str, bases: t.Tuple[type, ...], attrs: t.Dict[str, t. raise TypeError("all hooks must be an instance of ExecutionHook") options: t.Dict[str, options_.OptionData[t.Any]] = {} - invoke_method: t.Optional[str] = None + invoke_method: str | None = None # Iterate through new class attributes to find options and invoke method for name, item in attrs.items(): if cls._is_option(item): @@ -282,7 +282,7 @@ class CommandBase: __slots__ = ("_current_context", "_resolved_option_cache") _command_data: t.ClassVar[CommandData] - _current_context: t.Optional[context_.Context] + _current_context: context_.Context | None _resolved_option_cache: t.MutableMapping[str, t.Any] def __new__(cls, *args: t.Any, **kwargs: t.Any) -> CommandBase: @@ -304,7 +304,7 @@ def _set_context(self, context: context_.Context) -> None: self._current_context = context self._resolved_option_cache = {} - def _resolve_option(self, option: options_.Option[T, D]) -> t.Union[T, D]: + def _resolve_option(self, option: options_.Option[T, D]) -> T | D: """ Resolves the actual value for the given option from the command's current execution context. If the value has been resolved before and is available in the cache then diff --git a/lightbulb/commands/execution.py b/lightbulb/commands/execution.py index 09dbc995..b8d6f006 100644 --- a/lightbulb/commands/execution.py +++ b/lightbulb/commands/execution.py @@ -28,15 +28,14 @@ from lightbulb import exceptions from lightbulb.internal import constants from lightbulb.internal import di +from lightbulb.internal.types import MaybeAwaitable if t.TYPE_CHECKING: from lightbulb import context as context_ __all__ = ["ExecutionStep", "ExecutionSteps", "ExecutionHook", "ExecutionPipeline", "hook", "invoke"] -ExecutionHookFuncT: t.TypeAlias = t.Callable[ - ["ExecutionPipeline", "context_.Context"], t.Union[t.Awaitable[None], None] -] +ExecutionHookFuncT: t.TypeAlias = t.Callable[["ExecutionPipeline", "context_.Context"], MaybeAwaitable[None]] @dataclasses.dataclass(frozen=True, slots=True, eq=True) @@ -110,14 +109,14 @@ def __init__(self, context: context_.Context, order: t.Sequence[ExecutionStep]) self._context = context self._remaining = list(order) - self._hooks: t.Dict[ExecutionStep, t.List[ExecutionHook]] = collections.defaultdict(list) + self._hooks: dict[ExecutionStep, list[ExecutionHook]] = collections.defaultdict(list) for hook in context.command_data.hooks: self._hooks[hook.step].append(hook) - self._current_step: t.Optional[ExecutionStep] = None - self._current_hook: t.Optional[ExecutionHook] = None + self._current_step: ExecutionStep | None = None + self._current_hook: ExecutionHook | None = None - self._failure: t.Optional[exceptions.HookFailedException] = None + self._failure: exceptions.HookFailedException | None = None @property def failed(self) -> bool: @@ -128,7 +127,7 @@ def failed(self) -> bool: """ return self._failure is not None - def _next_step(self) -> t.Optional[ExecutionStep]: + def _next_step(self) -> ExecutionStep | None: """ Return the next execution step to run, or :obj:`None` if the remaining execution steps have been exhausted. @@ -178,7 +177,7 @@ async def _run(self) -> None: except Exception as e: raise exceptions.InvocationFailedException(e, self._context) - def fail(self, exc: t.Union[str, Exception]) -> None: + def fail(self, exc: str | Exception) -> None: """ Notify the pipeline of a failure in an execution hook. diff --git a/lightbulb/commands/groups.py b/lightbulb/commands/groups.py index 063965dc..01d5a5b3 100644 --- a/lightbulb/commands/groups.py +++ b/lightbulb/commands/groups.py @@ -34,9 +34,9 @@ from lightbulb import localization from lightbulb.commands import commands -CommandT = t.TypeVar("CommandT", bound=t.Type["commands.CommandBase"]) -SubGroupCommandMappingT = t.Dict[str, t.Type["commands.CommandBase"]] -GroupCommandMappingT = t.Dict[str, t.Union["SubGroup", t.Type["commands.CommandBase"]]] +CommandT = t.TypeVar("CommandT", bound=type["commands.CommandBase"]) +SubGroupCommandMappingT = dict[str, type["commands.CommandBase"]] +GroupCommandMappingT = dict[str, t.Union["SubGroup", type["commands.CommandBase"]]] class GroupMixin(abc.ABC): @@ -44,7 +44,7 @@ class GroupMixin(abc.ABC): __slots__ = () - _commands: t.Union[SubGroupCommandMappingT, GroupCommandMappingT] + _commands: SubGroupCommandMappingT | GroupCommandMappingT @t.overload def register(self) -> t.Callable[[CommandT], CommandT]: ... @@ -52,7 +52,7 @@ def register(self) -> t.Callable[[CommandT], CommandT]: ... @t.overload def register(self, command: CommandT) -> CommandT: ... - def register(self, command: t.Optional[CommandT] = None) -> t.Union[CommandT, t.Callable[[CommandT], CommandT]]: + def register(self, command: CommandT | None = None) -> CommandT | t.Callable[[CommandT], CommandT]: """ Register a command as a subcommand for this group. Can be used as a first or second order decorator, or called with the command to register. @@ -91,7 +91,7 @@ def _inner(_command: CommandT) -> CommandT: return _inner - def resolve_subcommand(self, path: t.List[str]) -> t.Optional[t.Type[commands.CommandBase]]: + def resolve_subcommand(self, path: list[str]) -> type[commands.CommandBase] | None: """ Resolve the subcommand for the given path - fully qualified command name. diff --git a/lightbulb/commands/options.py b/lightbulb/commands/options.py index 4d0434e2..f3c6c3c1 100644 --- a/lightbulb/commands/options.py +++ b/lightbulb/commands/options.py @@ -81,9 +81,9 @@ class OptionData(t.Generic[D]): channel_types: hikari.UndefinedOr[t.Sequence[hikari.ChannelType]] = hikari.UNDEFINED """The channel types for the option.""" - min_value: hikari.UndefinedOr[t.Union[int, float]] = hikari.UNDEFINED + min_value: hikari.UndefinedOr[int | float] = hikari.UNDEFINED """The minimum value for the option.""" - max_value: hikari.UndefinedOr[t.Union[int, float]] = hikari.UNDEFINED + max_value: hikari.UndefinedOr[int | float] = hikari.UNDEFINED """The maximum value for the option.""" min_length: hikari.UndefinedOr[int] = hikari.UNDEFINED @@ -176,7 +176,7 @@ def __init__(self, data: OptionData[D], default_when_not_bound: T) -> None: self._data = data self._unbound_default = default_when_not_bound - def __get__(self, instance: t.Optional[commands.CommandBase], owner: t.Type[commands.CommandBase]) -> t.Union[T, D]: + def __get__(self, instance: commands.CommandBase | None, owner: type[commands.CommandBase]) -> T | D: if instance is None or getattr(instance, "_current_context", None) is None: return self._unbound_default @@ -199,8 +199,8 @@ class ContextMenuOption(Option[CtxMenuOptionReturnT, CtxMenuOptionReturnT]): __slots__ = ("_type",) - def __init__(self, type: t.Type[CtxMenuOptionReturnT]) -> None: - self._type = type + def __init__(self, type_: type[CtxMenuOptionReturnT]) -> None: + self._type = type_ super().__init__( OptionData( type=hikari.OptionType.STRING, @@ -211,18 +211,14 @@ def __init__(self, type: t.Type[CtxMenuOptionReturnT]) -> None: ) @t.overload - def __get__( - self, instance: t.Optional[commands.UserCommand], owner: t.Type[commands.UserCommand] - ) -> hikari.User: ... + def __get__(self, instance: commands.UserCommand | None, owner: type[commands.UserCommand]) -> hikari.User: ... @t.overload def __get__( - self, instance: t.Optional[commands.MessageCommand], owner: t.Type[commands.MessageCommand] + self, instance: commands.MessageCommand | None, owner: type[commands.MessageCommand] ) -> hikari.Message: ... - def __get__( - self, instance: t.Optional[commands.CommandBase], owner: t.Type[commands.CommandBase] - ) -> CtxMenuOptionReturnT: + def __get__(self, instance: commands.CommandBase | None, owner: type[commands.CommandBase]) -> CtxMenuOptionReturnT: if instance is None or getattr(instance, "_current_context", None) is None: return self._unbound_default @@ -244,24 +240,16 @@ def __get__( # TODO - consider how to implement choice localisation def _normalise_choices( - choices: t.Union[ - t.Sequence[hikari.CommandChoice], - t.Mapping[str, t.Union[str, int, float]], - t.Sequence[t.Tuple[str, t.Union[str, int, float]]], - t.Sequence[t.Union[str, int, float]], - ], + choices: t.Sequence[hikari.CommandChoice] + | t.Mapping[str, str | int | float] + | t.Sequence[tuple[str, str | int | float]] + | t.Sequence[str | int | float], ) -> t.Sequence[hikari.CommandChoice]: if isinstance(choices, collections.abc.Mapping): return [hikari.CommandChoice(name=k, value=v) for k, v in choices.items()] def _to_command_choice( - item: t.Union[ - hikari.CommandChoice, - t.Tuple[str, t.Union[str, int, float]], - str, - int, - float, - ], + item: hikari.CommandChoice | tuple[str, str | int | float] | str | int | float, ) -> hikari.CommandChoice: if isinstance(item, hikari.CommandChoice): return item @@ -282,7 +270,7 @@ def string( localize: bool = False, default: hikari.UndefinedOr[D] = hikari.UNDEFINED, choices: hikari.UndefinedOr[ - t.Union[t.Sequence[hikari.CommandChoice], t.Mapping[str, str], t.Sequence[t.Tuple[str, str]], t.Sequence[str]] + t.Sequence[hikari.CommandChoice] | t.Mapping[str, str] | t.Sequence[tuple[str, str]] | t.Sequence[str] ] = hikari.UNDEFINED, min_length: hikari.UndefinedOr[int] = hikari.UNDEFINED, max_length: hikari.UndefinedOr[int] = hikari.UNDEFINED, @@ -341,7 +329,7 @@ def integer( localize: bool = False, default: hikari.UndefinedOr[D] = hikari.UNDEFINED, choices: hikari.UndefinedOr[ - t.Union[t.Sequence[hikari.CommandChoice], t.Mapping[str, int], t.Sequence[t.Tuple[str, int]], t.Sequence[int]] + t.Sequence[hikari.CommandChoice] | t.Mapping[str, int] | t.Sequence[t.Tuple[str, int]] | t.Sequence[int] ] = hikari.UNDEFINED, min_value: hikari.UndefinedOr[int] = hikari.UNDEFINED, max_value: hikari.UndefinedOr[int] = hikari.UNDEFINED, @@ -437,9 +425,7 @@ def number( localize: bool = False, default: hikari.UndefinedOr[D] = hikari.UNDEFINED, choices: hikari.UndefinedOr[ - t.Union[ - t.Sequence[hikari.CommandChoice], t.Mapping[str, float], t.Sequence[t.Tuple[str, float]], t.Sequence[float] - ] + t.Sequence[hikari.CommandChoice] | t.Mapping[str, float] | t.Sequence[tuple[str, float]] | t.Sequence[float] ] = hikari.UNDEFINED, min_value: hikari.UndefinedOr[float] = hikari.UNDEFINED, max_value: hikari.UndefinedOr[float] = hikari.UNDEFINED, diff --git a/lightbulb/commands/utils.py b/lightbulb/commands/utils.py index d9e0df49..387d9d2c 100644 --- a/lightbulb/commands/utils.py +++ b/lightbulb/commands/utils.py @@ -32,19 +32,19 @@ def localize_name_and_description( name: str, - description: t.Optional[str], + description: str | None, default_locale: hikari.Locale, localization_provider: localization.LocalizationProviderT, -) -> t.Tuple[str, str, t.Mapping[hikari.Locale, str], t.Mapping[hikari.Locale, str]]: +) -> tuple[str, str, t.Mapping[hikari.Locale, str], t.Mapping[hikari.Locale, str]]: name_localizations: t.Mapping[hikari.Locale, str] = localization_provider(name) - localized_name: t.Optional[str] = name_localizations.get(default_locale, None) + localized_name: str | None = name_localizations.get(default_locale, None) if localized_name is None: raise exceptions.LocalizationFailedException(f"failed to resolve key {name!r} for default locale") description_localizations: t.Mapping[hikari.Locale, str] = ( {} if description is None else localization_provider(description) ) - localized_description: t.Optional[str] = ( + localized_description: str | None = ( "" if description is None else description_localizations.get(default_locale, None) ) diff --git a/lightbulb/context.py b/lightbulb/context.py index 87cb5d5d..1eecf100 100644 --- a/lightbulb/context.py +++ b/lightbulb/context.py @@ -37,17 +37,17 @@ from lightbulb import client as client_ from lightbulb import commands -AutocompleteResponseT = t.Union[ +AutocompleteResponse: t.TypeAlias = t.Union[ t.Sequence[special_endpoints.AutocompleteChoiceBuilder], t.Sequence[str], t.Mapping[str, str], - t.Sequence[t.Tuple[str, str]], + t.Sequence[tuple[str, str]], t.Sequence[int], t.Mapping[str, int], - t.Sequence[t.Tuple[str, int]], + t.Sequence[tuple[str, int]], t.Sequence[float], t.Mapping[str, float], - t.Sequence[t.Tuple[str, float]], + t.Sequence[tuple[str, float]], ] INITIAL_RESPONSE_IDENTIFIER: t.Final[int] = -1 @@ -63,10 +63,10 @@ class AutocompleteContext: options: t.Sequence[hikari.AutocompleteInteractionOption] """The options provided with the autocomplete interaction.""" - command: t.Type[commands.CommandBase] + command: type[commands.CommandBase] """Command class for the autocomplete invocation.""" - _focused: t.Optional[hikari.AutocompleteInteractionOption] = dataclasses.field(init=False, default=None) + _focused: hikari.AutocompleteInteractionOption | None = dataclasses.field(init=False, default=None) @property def focused(self) -> hikari.AutocompleteInteractionOption: @@ -84,7 +84,7 @@ def focused(self) -> hikari.AutocompleteInteractionOption: self._focused = found return self._focused - def get_option(self, name: str) -> t.Optional[hikari.AutocompleteInteractionOption]: + def get_option(self, name: str) -> hikari.AutocompleteInteractionOption | None: """ Get the option with the given name if available. @@ -101,16 +101,16 @@ def get_option(self, name: str) -> t.Optional[hikari.AutocompleteInteractionOpti return next(filter(lambda opt: opt.name == name, self.options), None) @staticmethod - def _normalise_choices(choices: AutocompleteResponseT) -> t.Sequence[special_endpoints.AutocompleteChoiceBuilder]: + def _normalise_choices(choices: AutocompleteResponse) -> t.Sequence[special_endpoints.AutocompleteChoiceBuilder]: if isinstance(choices, collections.abc.Mapping): return [hikari.impl.AutocompleteChoiceBuilder(name=k, value=v) for k, v in choices.items()] def _to_command_choice( item: t.Union[ special_endpoints.AutocompleteChoiceBuilder, - t.Tuple[str, str], - t.Tuple[str, int], - t.Tuple[str, float], + tuple[str, str], + tuple[str, int], + tuple[str, float], str, int, float, @@ -126,7 +126,7 @@ def _to_command_choice( return list(map(_to_command_choice, choices)) - async def respond(self, choices: AutocompleteResponseT) -> None: + async def respond(self, choices: AutocompleteResponse) -> None: """ Create a response for the autocomplete interaction this context represents. @@ -147,7 +147,7 @@ class RestAutocompleteContext(AutocompleteContext): None, ] - async def respond(self, choices: AutocompleteResponseT) -> None: + async def respond(self, choices: AutocompleteResponse) -> None: normalised_choices = self._normalise_choices(choices) self._initial_response_callback(special_endpoints_impl.InteractionAutocompleteBuilder(normalised_choices)) @@ -174,7 +174,7 @@ def __post_init__(self) -> None: self.command._set_context(self) @property - def guild_id(self) -> t.Optional[hikari.Snowflake]: + def guild_id(self) -> hikari.Snowflake | None: """The ID of the guild that the command was invoked in. :obj:`None` if the invocation occurred in DM.""" return self.interaction.guild_id @@ -189,7 +189,7 @@ def user(self) -> hikari.User: return self.interaction.user @property - def member(self) -> t.Optional[hikari.InteractionMember]: + def member(self) -> hikari.InteractionMember | None: """The member that invoked the command, if it was invoked in a guild.""" return self.interaction.member @@ -203,21 +203,15 @@ async def edit_response( response_id: hikari.Snowflakeish, content: hikari.UndefinedNoneOr[t.Any] = hikari.UNDEFINED, *, - attachment: hikari.UndefinedNoneOr[t.Union[hikari.Resourceish, hikari.Attachment]] = hikari.UNDEFINED, - attachments: hikari.UndefinedNoneOr[ - t.Sequence[t.Union[hikari.Resourceish, hikari.Attachment]] - ] = hikari.UNDEFINED, + attachment: hikari.UndefinedNoneOr[hikari.Resourceish | hikari.Attachment] = hikari.UNDEFINED, + attachments: hikari.UndefinedNoneOr[t.Sequence[hikari.Resourceish | hikari.Attachment]] = hikari.UNDEFINED, component: hikari.UndefinedNoneOr[special_endpoints.ComponentBuilder] = hikari.UNDEFINED, components: hikari.UndefinedNoneOr[t.Sequence[special_endpoints.ComponentBuilder]] = hikari.UNDEFINED, embed: hikari.UndefinedNoneOr[hikari.Embed] = hikari.UNDEFINED, embeds: hikari.UndefinedNoneOr[t.Sequence[hikari.Embed]] = hikari.UNDEFINED, mentions_everyone: hikari.UndefinedOr[bool] = hikari.UNDEFINED, - user_mentions: hikari.UndefinedOr[ - t.Union[hikari.SnowflakeishSequence[hikari.PartialUser], bool] - ] = hikari.UNDEFINED, - role_mentions: hikari.UndefinedOr[ - t.Union[hikari.SnowflakeishSequence[hikari.PartialRole], bool] - ] = hikari.UNDEFINED, + user_mentions: hikari.UndefinedOr[hikari.SnowflakeishSequence[hikari.PartialUser] | bool] = hikari.UNDEFINED, + role_mentions: hikari.UndefinedOr[hikari.SnowflakeishSequence[hikari.PartialRole] | bool] = hikari.UNDEFINED, ) -> hikari.Message: """ Edit the response with the given identifier. @@ -307,7 +301,7 @@ async def _create_initial_response( response_type: t.Literal[hikari.ResponseType.MESSAGE_CREATE, hikari.ResponseType.DEFERRED_MESSAGE_CREATE], content: hikari.UndefinedOr[t.Any] = hikari.UNDEFINED, *, - flags: t.Union[int, hikari.MessageFlag, hikari.UndefinedType] = hikari.UNDEFINED, + flags: int | hikari.MessageFlag | hikari.UndefinedType = hikari.UNDEFINED, tts: hikari.UndefinedOr[bool] = hikari.UNDEFINED, attachment: hikari.UndefinedOr[hikari.Resourceish] = hikari.UNDEFINED, attachments: hikari.UndefinedOr[t.Sequence[hikari.Resourceish]] = hikari.UNDEFINED, @@ -316,12 +310,8 @@ async def _create_initial_response( embed: hikari.UndefinedOr[hikari.Embed] = hikari.UNDEFINED, embeds: hikari.UndefinedOr[t.Sequence[hikari.Embed]] = hikari.UNDEFINED, mentions_everyone: hikari.UndefinedOr[bool] = hikari.UNDEFINED, - user_mentions: hikari.UndefinedOr[ - t.Union[hikari.SnowflakeishSequence[hikari.PartialUser], bool] - ] = hikari.UNDEFINED, - role_mentions: hikari.UndefinedOr[ - t.Union[hikari.SnowflakeishSequence[hikari.PartialRole], bool] - ] = hikari.UNDEFINED, + user_mentions: hikari.UndefinedOr[hikari.SnowflakeishSequence[hikari.PartialUser] | bool] = hikari.UNDEFINED, + role_mentions: hikari.UndefinedOr[hikari.SnowflakeishSequence[hikari.PartialRole] | bool] = hikari.UNDEFINED, ) -> hikari.Snowflakeish: await self.interaction.create_initial_response( response_type, @@ -393,7 +383,7 @@ async def respond( self, content: hikari.UndefinedOr[t.Any] = hikari.UNDEFINED, *, - flags: t.Union[int, hikari.MessageFlag, hikari.UndefinedType] = hikari.UNDEFINED, + flags: int | hikari.MessageFlag | hikari.UndefinedType = hikari.UNDEFINED, tts: hikari.UndefinedOr[bool] = hikari.UNDEFINED, attachment: hikari.UndefinedOr[hikari.Resourceish] = hikari.UNDEFINED, attachments: hikari.UndefinedOr[t.Sequence[hikari.Resourceish]] = hikari.UNDEFINED, @@ -402,12 +392,8 @@ async def respond( embed: hikari.UndefinedOr[hikari.Embed] = hikari.UNDEFINED, embeds: hikari.UndefinedOr[t.Sequence[hikari.Embed]] = hikari.UNDEFINED, mentions_everyone: hikari.UndefinedOr[bool] = hikari.UNDEFINED, - user_mentions: hikari.UndefinedOr[ - t.Union[hikari.SnowflakeishSequence[hikari.PartialUser], bool] - ] = hikari.UNDEFINED, - role_mentions: hikari.UndefinedOr[ - t.Union[hikari.SnowflakeishSequence[hikari.PartialRole], bool] - ] = hikari.UNDEFINED, + user_mentions: hikari.UndefinedOr[hikari.SnowflakeishSequence[hikari.PartialUser] | bool] = hikari.UNDEFINED, + role_mentions: hikari.UndefinedOr[hikari.SnowflakeishSequence[hikari.PartialRole] | bool] = hikari.UNDEFINED, ) -> hikari.Snowflakeish: """ Create a response to the interaction that this context represents. @@ -516,7 +502,7 @@ async def _create_initial_response( response_type: hikari.ResponseType, content: hikari.UndefinedOr[t.Any] = hikari.UNDEFINED, *, - flags: t.Union[int, hikari.MessageFlag, hikari.UndefinedType] = hikari.UNDEFINED, + flags: int | hikari.MessageFlag | hikari.UndefinedType = hikari.UNDEFINED, tts: hikari.UndefinedOr[bool] = hikari.UNDEFINED, attachment: hikari.UndefinedOr[hikari.Resourceish] = hikari.UNDEFINED, attachments: hikari.UndefinedOr[t.Sequence[hikari.Resourceish]] = hikari.UNDEFINED, @@ -525,12 +511,8 @@ async def _create_initial_response( embed: hikari.UndefinedOr[hikari.Embed] = hikari.UNDEFINED, embeds: hikari.UndefinedOr[t.Sequence[hikari.Embed]] = hikari.UNDEFINED, mentions_everyone: hikari.UndefinedOr[bool] = hikari.UNDEFINED, - user_mentions: hikari.UndefinedOr[ - t.Union[hikari.SnowflakeishSequence[hikari.PartialUser], bool] - ] = hikari.UNDEFINED, - role_mentions: hikari.UndefinedOr[ - t.Union[hikari.SnowflakeishSequence[hikari.PartialRole], bool] - ] = hikari.UNDEFINED, + user_mentions: hikari.UndefinedOr[hikari.SnowflakeishSequence[hikari.PartialUser] | bool] = hikari.UNDEFINED, + role_mentions: hikari.UndefinedOr[hikari.SnowflakeishSequence[hikari.PartialRole] | bool] = hikari.UNDEFINED, ) -> hikari.Snowflakeish: if attachment and attachments: raise ValueError("You may only specify one of 'attachment' or 'attachments', not both") diff --git a/lightbulb/internal/di.py b/lightbulb/internal/di.py index f69156cc..a1ca17eb 100644 --- a/lightbulb/internal/di.py +++ b/lightbulb/internal/di.py @@ -25,6 +25,9 @@ import svcs +if t.TYPE_CHECKING: + from lightbulb.internal.types import MaybeAwaitable + T = t.TypeVar("T") AnyAsyncCallableT = t.TypeVar("AnyAsyncCallableT", bound=t.Callable[..., t.Awaitable[t.Any]]) @@ -40,7 +43,7 @@ class DependencyInjectionManager: def __init__(self) -> None: self._di_registry: svcs.Registry = svcs.Registry() - self._di_container: t.Optional[svcs.Container] = None + self._di_container: svcs.Container | None = None @property def di_registry(self) -> svcs.Registry: @@ -54,19 +57,19 @@ def di_container(self) -> svcs.Container: self._di_container = svcs.Container(self.di_registry) return self._di_container - def register_dependency(self, type: t.Type[T], factory: t.Callable[[], t.Union[t.Awaitable[T], T]]) -> None: + def register_dependency(self, type_: type[T], factory: t.Callable[[], MaybeAwaitable[T]]) -> None: """ Register a dependency as usable by dependency injection. All dependencies are considered to be singletons, meaning the factory will always be called at most once. Args: - type (:obj:`~typing.Type` [ ``T`` ]): The type of the dependency to register. + type_ (:obj:`~typing.Type` [ ``T`` ]): The type of the dependency to register. factory: The factory function to use to provide the dependency value. Returns: :obj:`None` """ - self.di_registry.register_factory(type, factory) # type: ignore[reportUnknownMemberType] + self.di_registry.register_factory(type_, factory) # type: ignore[reportUnknownMemberType] @contextlib.contextmanager @@ -97,7 +100,7 @@ def ensure_di_context(client: DependencyInjectionManager) -> t.Generator[None, t def find_injectable_kwargs( func: t.Callable[..., t.Any], passed_args: int, passed_kwargs: t.Collection[str] -) -> t.Dict[str, t.Any]: +) -> dict[str, t.Any]: """ Given a function, parse the signature to discover which parameters are suitable for dependency injection. @@ -121,7 +124,7 @@ def find_injectable_kwargs( """ parameters = inspect.signature(func, eval_str=True).parameters - injectable_parameters: t.Dict[str, t.Any] = {} + injectable_parameters: dict[str, t.Any] = {} for parameter in [*parameters.values()][passed_args:]: # Injectable parameters MUST have an annotation and no default if ( @@ -163,7 +166,7 @@ def __init__( self._func = func self._self: t.Any = self_ - def __get__(self, instance: t.Any, owner: t.Type[t.Any]) -> LazyInjecting: + def __get__(self, instance: t.Any, owner: type[t.Any]) -> LazyInjecting: if instance is not None: return LazyInjecting(self._func, instance) return self @@ -175,7 +178,7 @@ def __setattr__(self, key: str, value: t.Any) -> None: setattr(self._func, key, value) async def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any: - new_kwargs: t.Dict[str, t.Any] = {} + new_kwargs: dict[str, t.Any] = {} new_kwargs.update(kwargs) di_container: t.Optional[svcs.Container] = DI_CONTAINER.get(None) diff --git a/lightbulb/internal/sync.py b/lightbulb/internal/sync.py index 4cc2d55d..9577ad00 100644 --- a/lightbulb/internal/sync.py +++ b/lightbulb/internal/sync.py @@ -37,9 +37,9 @@ @dataclasses.dataclass(slots=True) class CommandBuilderCollection: - slash: t.Optional[hikari.api.SlashCommandBuilder] = None - user: t.Optional[hikari.api.ContextMenuCommandBuilder] = None - message: t.Optional[hikari.api.ContextMenuCommandBuilder] = None + slash: hikari.api.SlashCommandBuilder | None = None + user: hikari.api.ContextMenuCommandBuilder | None = None + message: hikari.api.ContextMenuCommandBuilder | None = None def put(self, bld: hikari.api.CommandBuilder) -> None: if isinstance(bld, hikari.api.SlashCommandBuilder): @@ -55,8 +55,8 @@ def put(self, bld: hikari.api.CommandBuilder) -> None: def hikari_command_to_builder( cmd: hikari.PartialCommand, -) -> t.Union[hikari.api.SlashCommandBuilder, hikari.api.ContextMenuCommandBuilder]: - bld: t.Union[hikari.api.SlashCommandBuilder, hikari.api.ContextMenuCommandBuilder] +) -> hikari.api.SlashCommandBuilder | hikari.api.ContextMenuCommandBuilder: + bld: hikari.api.SlashCommandBuilder | hikari.api.ContextMenuCommandBuilder if desc := getattr(cmd, "description", None): bld = hikari.impl.SlashCommandBuilder(cmd.name, description=desc) for option in getattr(cmd, "options", []) or []: @@ -75,9 +75,9 @@ def hikari_command_to_builder( async def get_existing_and_registered_commands( client: client_.Client, application: hikari.PartialApplication, guild: hikari.UndefinedOr[hikari.Snowflakeish] -) -> t.Tuple[t.Dict[str, CommandBuilderCollection], t.Dict[str, CommandBuilderCollection]]: - existing: t.Dict[str, CommandBuilderCollection] = collections.defaultdict(CommandBuilderCollection) - registered: t.Dict[str, CommandBuilderCollection] = collections.defaultdict(CommandBuilderCollection) +) -> tuple[dict[str, CommandBuilderCollection], dict[str, CommandBuilderCollection]]: + existing: dict[str, CommandBuilderCollection] = collections.defaultdict(CommandBuilderCollection) + registered: dict[str, CommandBuilderCollection] = collections.defaultdict(CommandBuilderCollection) for existing_command in await client.rest.fetch_application_commands(application, guild=guild): existing[existing_command.name].put(hikari_command_to_builder(existing_command)) @@ -93,8 +93,8 @@ async def get_existing_and_registered_commands( return existing, registered -def serialize_builder(bld: hikari.api.CommandBuilder) -> t.Dict[str, t.Any]: - def serialize_option(opt: hikari.CommandOption) -> t.Dict[str, t.Any]: +def serialize_builder(bld: hikari.api.CommandBuilder) -> dict[str, t.Any]: + def serialize_option(opt: hikari.CommandOption) -> dict[str, t.Any]: return { "type": opt.type, "name": opt.name, @@ -112,7 +112,7 @@ def serialize_option(opt: hikari.CommandOption) -> t.Dict[str, t.Any]: "max_length": opt.max_length, } - out: t.Dict[str, t.Any] = { + out: dict[str, t.Any] = { "name": bld.name, "is_dm_enabled": non_undefined_or(bld.is_dm_enabled, True), "is_nsfw": non_undefined_or(bld.is_nsfw, False), @@ -128,13 +128,13 @@ def serialize_option(opt: hikari.CommandOption) -> t.Dict[str, t.Any]: def get_commands_to_set( - existing: t.Dict[str, CommandBuilderCollection], - registered: t.Dict[str, CommandBuilderCollection], + existing: dict[str, CommandBuilderCollection], + registered: dict[str, CommandBuilderCollection], delete_unknown: bool, -) -> t.Optional[t.Sequence[hikari.api.CommandBuilder]]: +) -> t.Sequence[hikari.api.CommandBuilder] | None: created, deleted, updated, unchanged = 0, 0, 0, 0 - commands_to_set: t.List[hikari.api.CommandBuilder] = [] + commands_to_set: list[hikari.api.CommandBuilder] = [] for name in {*existing.keys(), *registered.keys()}: existing_cmds, registered_cmds = existing[name], registered[name] for existing_bld, registered_bld in zip( diff --git a/lightbulb/internal/types.py b/lightbulb/internal/types.py new file mode 100644 index 00000000..06fe3201 --- /dev/null +++ b/lightbulb/internal/types.py @@ -0,0 +1,22 @@ +# -*- coding: utf-8 -*- +# Copyright © tandemdude 2023-present +# +# This file is part of Lightbulb. +# +# Lightbulb is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Lightbulb is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License +# along with Lightbulb. If not, see . +import typing as t + +T = t.TypeVar("T") + +MaybeAwaitable: t.TypeAlias = t.Union[T, t.Awaitable[T]] diff --git a/lightbulb/internal/utils.py b/lightbulb/internal/utils.py index 5ec86686..8f8348fc 100644 --- a/lightbulb/internal/utils.py +++ b/lightbulb/internal/utils.py @@ -31,13 +31,13 @@ @dataclasses.dataclass(slots=True) class CommandCollection: - slash: t.Optional[t.Union[groups.Group, t.Type[commands.SlashCommand]]] = None - user: t.Optional[t.Type[commands.UserCommand]] = None - message: t.Optional[t.Type[commands.MessageCommand]] = None + slash: groups.Group | type[commands.SlashCommand] | None = None + user: type[commands.UserCommand] | None = None + message: type[commands.MessageCommand] | None = None def put( self, - command: t.Union[groups.Group, t.Type[commands.CommandBase]], + command: groups.Group | t.Type[commands.CommandBase], ) -> None: if isinstance(command, groups.Group) or issubclass(command, commands.SlashCommand): self.slash = command @@ -49,5 +49,5 @@ def put( raise TypeError("unsupported command passed") -def non_undefined_or(item: hikari.UndefinedOr[T], default: D) -> t.Union[T, D]: +def non_undefined_or(item: hikari.UndefinedOr[T], default: D) -> T | D: return item if item is not hikari.UNDEFINED else default diff --git a/lightbulb/localization.py b/lightbulb/localization.py index ae9144c1..cac39359 100644 --- a/lightbulb/localization.py +++ b/lightbulb/localization.py @@ -67,7 +67,7 @@ class DictLocalizationProvider: """Mapping containing the localizations that can be provided.""" def __call__(self, key: str) -> LocalizationMappingT: - out: t.Dict[hikari.Locale, str] = {} + out: dict[hikari.Locale, str] = {} for locale, translations in self.localizations.items(): if key in translations: out[locale] = translations[key] @@ -102,7 +102,7 @@ def __post_init__(self) -> None: if not self.filename.endswith(".po") and not self.filename.endswith(".mo"): raise ValueError("'filename' - file must be of type '.po' or '.mo'") - localizations: t.Dict[hikari.Locale, t.Dict[str, str]] = collections.defaultdict(dict) + localizations: dict[hikari.Locale, dict[str, str]] = collections.defaultdict(dict) for directory in pathlib.Path(self.directory).iterdir(): if not directory.is_dir(): @@ -116,7 +116,7 @@ def __post_init__(self) -> None: if not translations_file.is_file(): continue - parsed: t.Union[polib.POFile, polib.MOFile] = ( + parsed: polib.POFile | polib.MOFile = ( polib.pofile(translations_file.as_posix()) if translations_file.name.endswith(".po") else polib.mofile(translations_file.as_posix()) diff --git a/lightbulb/utils.py b/lightbulb/utils.py index 22ebfc7c..fe4a4a10 100644 --- a/lightbulb/utils.py +++ b/lightbulb/utils.py @@ -126,7 +126,7 @@ class _EmptyUser(hikari.User): """Placeholder for an attachment. Used when attempting to get value for an option on a class instead of instance.""" -def get_command_data(command: t.Union[commands.CommandBase, t.Type[commands.CommandBase]]) -> commands.CommandData: +def get_command_data(command: commands.CommandBase | type[commands.CommandBase]) -> commands.CommandData: """ Utility method to get the command data dataclass for a command instance or command class.