Implement autocomplete

Author: CodeWithSwastik

discord/bot.py

@@ -59,6 +59,7 @@
 
 from .errors import Forbidden, DiscordException
 from .interactions import Interaction
+from .enums import InteractionType
 
 CoroFunc = Callable[..., Coroutine[Any, Any, Any]]
 CFT = TypeVar('CFT', bound=CoroFunc)
@@ -86,7 +87,7 @@ def pending_application_commands(self):
         return self._pending_application_commands
 
     @property
-    def commands(self) -> List[Union[ApplicationCommand, ...]]:
+    def commands(self) -> List[Union[ApplicationCommand, Any]]:
         commands = list(self.application_commands.values())
         if self._supports_prefixed_commands:
             commands += self.prefixed_commands
@@ -364,14 +365,20 @@ async def process_application_commands(self, interaction: Interaction) -> None:
         interaction: :class:`discord.Interaction`
             The interaction to process
         """
-        if not interaction.is_command():
+        if interaction.type not in (
+            InteractionType.application_command, 
+            InteractionType.auto_complete
+        ):
             return
 
         try:
             command = self.application_commands[interaction.data["id"]]
         except KeyError:
             self.dispatch("unknown_command", interaction)
         else:
+            if interaction.type is InteractionType.auto_complete:
+                return await command.invoke_autocomplete_callback(interaction)
+            
             ctx = await self.get_application_context(interaction)
             ctx.command = command
             self.dispatch("application_command", ctx)

discord/commands/commands.py

@@ -30,7 +30,7 @@
 import functools
 import inspect
 from collections import OrderedDict
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any, Callable, Dict, List, Optional, Union, TYPE_CHECKING
 
 from ..enums import SlashCommandOptionType, ChannelType
 from ..member import Member
@@ -60,6 +60,9 @@
     "MessageCommand",
 )
 
+if TYPE_CHECKING: 
+    from ..interactions import Interaction
+
 def wrap_callback(coro):
     @functools.wraps(coro)
     async def wrapped(*args, **kwargs):
@@ -351,7 +354,7 @@ def __init__(self, func: Callable, *args, **kwargs) -> None:
         self.cog = None
 
         params = self._get_signature_parameters()
-        self.options = self._parse_options(params)
+        self.options: List[Option] = self._parse_options(params)
 
         try:
             checks = func.__commands_checks__
@@ -487,6 +490,17 @@ async def _invoke(self, ctx: ApplicationContext) -> None:
         else:
             await self.callback(ctx, **kwargs)
 
+    async def invoke_autocomplete_callback(self, interaction: Interaction):
+        for op in interaction.data.get("options", []):
+            if op.get("focused", False):
+                option = find(lambda o: o.name == op["name"], self.options)
+                result = await option.autocomplete(interaction, op.get("value", None))
+                choices = [
+                    o if isinstance(o, OptionChoice) else OptionChoice(o)
+                    for o in result
+                ]
+                await interaction.response.send_autocomplete_result(choices=choices)
+
     def qualified_name(self):
         return self.name
 
@@ -581,13 +595,21 @@ def __init__(
         if not (isinstance(self.max_value, minmax_types) or self.min_value is None):
             raise TypeError(f"Expected {minmax_typehint} for max_value, got \"{type(self.max_value).__name__}\"")
 
+        self.autocomplete = kwargs.pop("autocomplete", None)
+        if (
+            self.autocomplete and 
+            not asyncio.iscoroutinefunction(self.autocomplete)
+        ):
+            raise TypeError("Autocomplete callback must be a coroutine.")
+
     def to_dict(self) -> Dict:
         as_dict = {
             "name": self.name,
             "description": self.description,
             "type": self.input_type.value,
             "required": self.required,
             "choices": [c.to_dict() for c in self.choices],
+            "autocomplete": bool(self.autocomplete)
         }
         if self.channel_types:
             as_dict["channel_types"] = [t.value for t in self.channel_types]
@@ -722,6 +744,13 @@ async def _invoke(self, ctx: ApplicationContext) -> None:
         ctx.interaction.data = option
         await command.invoke(ctx)
 
+    async def invoke_autocomplete_callback(self, interaction: Interaction) -> None:
+        option = interaction.data["options"][0]
+        command = find(lambda x: x.name == option["name"], self.subcommands)
+        interaction.data = option
+        await command.invoke_autocomplete_callback(interaction)
+
+
 class ContextMenuCommand(ApplicationCommand):
     r"""A class that implements the protocol for context menu commands.
 

discord/interactions.py

@@ -61,11 +61,13 @@
     from .ui.view import View
     from .channel import VoiceChannel, StageChannel, TextChannel, CategoryChannel, StoreChannel, PartialMessageable
     from .threads import Thread
+    from .commands import OptionChoice
 
     InteractionChannel = Union[
         VoiceChannel, StageChannel, TextChannel, CategoryChannel, StoreChannel, Thread, PartialMessageable
     ]
 
+
 MISSING: Any = utils.MISSING
 
 
@@ -707,7 +709,47 @@ async def edit_message(
 
         self._responded = True
 
+    async def send_autocomplete_result(
+        self,
+        *,
+        choices: List[OptionChoice],
+    ) -> None:
+        """|coro|
+        Responds to this interaction by sending the autocomplete choices.
+        Parameters
+        -----------
+        choices: List[:class:`OptionChoice`]
+            A list of choices.  
+        Raises
+        -------
+        HTTPException
+            Sending the result failed.
+        InteractionResponded
+            This interaction has already been responded to before.
+        """
+        if self._responded:
+            raise InteractionResponded(self._parent)
+
+        parent = self._parent
+
+        if parent.type is not InteractionType.auto_complete:
+            return
+
+        payload = {
+            "choices": [c.to_dict() for c in choices]
+        }
+
+        adapter = async_context.get()
+        await adapter.create_interaction_response(
+            parent.id,
+            parent.token,
+            session=parent._session,
+            type=InteractionResponseType.auto_complete_result.value,
+            data=payload,
+        )
 
+        self._responded = True
+        
 class _InteractionMessageState:
     __slots__ = ('_parent', '_interaction')