From e6c8bf1bff674a7c5c36b392f58a20f8042c4087 Mon Sep 17 00:00:00 2001
From: Antonis Geralis <43617260+planetis-m@users.noreply.github.com>
Date: Mon, 23 Sep 2024 20:00:38 +0300
Subject: [PATCH] Added a Completer that throttles superfluous calls

---
 src/prompt_toolkit/completion/base.py | 41 +++++++++++++++++++++++++++
 1 file changed, 41 insertions(+)

diff --git a/src/prompt_toolkit/completion/base.py b/src/prompt_toolkit/completion/base.py
index 3846ef756..30a2c00c9 100644
--- a/src/prompt_toolkit/completion/base.py
+++ b/src/prompt_toolkit/completion/base.py
@@ -16,6 +16,7 @@
     "ThreadedCompleter",
     "DummyCompleter",
     "DynamicCompleter",
+    "ThrottledCompleter",
     "CompleteEvent",
     "ConditionalCompleter",
     "merge_completers",
@@ -316,6 +317,46 @@ def __repr__(self) -> str:
         return f"DynamicCompleter({self.get_completer!r} -> {self.get_completer()!r})"
 
 
+class ThrottledCompleter(Completer):
+    """
+    Completer that throttles completion requests to prevent rapid successive calls.
+
+    :param completer: :class:`.Completer` instance.
+    :param delay: Delay in seconds between completions. Default is 0.3.
+    """
+
+    def __init__(self, completer: Completer, delay: float = 0.3):
+        self.completer = completer
+        self.delay = delay
+        self._last_completion_time = 0
+        self._semaphore = asyncio.Semaphore(1)
+
+    def get_completions(
+        self, document: Document, complete_event: CompleteEvent
+    ) -> Iterable[Completion]:
+        """Get completions without throttling."""
+        yield from self.completer.get_completions(document, complete_event)
+
+    async def get_completions_async(
+        self, document: Document, complete_event: CompleteEvent
+    ) -> AsyncGenerator[Completion, None]:
+        """Get completions asynchronously, applying throttle delay."""
+        async with self._semaphore:
+            current_time = asyncio.get_running_loop().time()
+            time_since_last_completion = current_time - self._last_completion_time
+
+            if time_since_last_completion < self.delay:
+                await asyncio.sleep(self.delay - time_since_last_completion)
+
+            self._last_completion_time = asyncio.get_running_loop().time()
+
+            async with aclosing(
+                self.completer.get_completions_async(document, complete_event)
+            ) as async_generator:
+                async for item in async_generator:
+                    yield item
+
+
 class ConditionalCompleter(Completer):
     """
     Wrapper around any other completer that will enable/disable the completions