Customized concurrent handling

docs/source/telegram.ext.BaseProcessor.rst

@@ -0,0 +1,6 @@
+BaseProcessor
+======================
+
+.. autoclass:: telegram.ext.BaseProcessor
+    :members:
+    :show-inheritance:

telegram/ext/__init__.py

@@ -25,6 +25,7 @@
     "ApplicationHandlerStop",
     "BaseHandler",
     "BasePersistence",
+    "BaseProcessor",
     "BaseRateLimiter",
     "CallbackContext",
     "CallbackDataCache",
@@ -59,7 +60,7 @@
 
 from . import filters
 from ._aioratelimiter import AIORateLimiter
-from ._application import Application, ApplicationHandlerStop
+from ._application import Application, ApplicationHandlerStop, BaseProcessor
 from ._applicationbuilder import ApplicationBuilder
 from ._basepersistence import BasePersistence, PersistenceInput
 from ._baseratelimiter import BaseRateLimiter

telegram/ext/_application.py

@@ -24,6 +24,8 @@
 import logging
 import platform
 import signal
+from abc import abstractmethod
+from asyncio import BoundedSemaphore
 from collections import defaultdict
 from copy import deepcopy
 from pathlib import Path
@@ -259,7 +261,7 @@ def __init__(
         update_queue: "asyncio.Queue[object]",
         updater: Optional[Updater],
         job_queue: JQ,
-        concurrent_updates: Union[bool, int],
+        concurrent_updates: Union[bool, int, "BaseProcessor"],
         persistence: Optional[BasePersistence[UD, CD, BD]],
         context_types: ContextTypes[CCT, UD, CD, BD],
         post_init: Optional[
@@ -297,14 +299,16 @@ def __init__(
         self.post_stop: Optional[
             Callable[["Application[BT, CCT, UD, CD, BD, JQ]"], Coroutine[Any, Any, None]]
         ] = post_stop
-
-        if isinstance(concurrent_updates, int) and concurrent_updates < 0:
-            raise ValueError("`concurrent_updates` must be a non-negative integer!")
         if concurrent_updates is True:
             concurrent_updates = 256
-        self._concurrent_updates_sem = asyncio.BoundedSemaphore(concurrent_updates or 1)
-        self._concurrent_updates: int = concurrent_updates or 0
-
+        if isinstance(concurrent_updates, int):
+            if concurrent_updates < 0:
+                raise ValueError("`concurrent_updates` must be a non-negative integer!")
+            concurrent_updates = SimpleProcessor(concurrent_updates)
+        self._concurrent_updates_sem = asyncio.BoundedSemaphore(
+            concurrent_updates.max_concurrent_updates or 1
+        )
+        self._concurrent_updates: BaseProcessor = concurrent_updates
         self.bot_data: BD = self.context_types.bot_data()
         self._user_data: DefaultDict[int, UD] = defaultdict(self.context_types.user_data)
         self._chat_data: DefaultDict[int, CD] = defaultdict(self.context_types.chat_data)
@@ -361,7 +365,7 @@ def concurrent_updates(self) -> int:
 
         .. seealso:: :wiki:`Concurrency`
         """
-        return self._concurrent_updates
+        return self._concurrent_updates.max_concurrent_updates
 
     @property
     def job_queue(self) -> Optional["JobQueue[CCT]"]:
@@ -379,6 +383,14 @@ def job_queue(self) -> Optional["JobQueue[CCT]"]:
             )
         return self._job_queue
 
+    @property
+    def processor(self) -> Optional["BaseProcessor"]:
+        """:class:`telegram.ext.BaseProcessor`: The processor used by this application.
+
+        .. seealso:: :wiki:`Concurrency`
+        """
+        return self._concurrent_updates
+
     async def initialize(self) -> None:
         """Initializes the Application by initializing:
 
@@ -1047,7 +1059,6 @@ async def _update_fetcher(self) -> None:
         # Continuously fetch updates from the queue. Exit only once the signal object is found.
         while True:
             update = await self.update_queue.get()
-
             if update is _STOP_SIGNAL:
                 _logger.debug("Dropping pending updates")
                 while not self.update_queue.empty():
@@ -1061,7 +1072,12 @@ async def _update_fetcher(self) -> None:
 
             if self._concurrent_updates:
                 # We don't await the below because it has to be run concurrently
-                self.create_task(self.__process_update_wrapper(update), update=update)
+                self.create_task(
+                    self._concurrent_updates.do_process_update(
+                        update, self.process_update(update)
+                    ),
+                    update=update,
+                )
             else:
                 await self.__process_update_wrapper(update)
 
@@ -1672,3 +1688,64 @@ async def process_error(
 
         _logger.exception("No error handlers are registered, logging exception.", exc_info=error)
         return False
+
+
+class BaseProcessor:
+    def __init__(self, max_concurrent_updates: int):
+        self.max_concurrent_updates = max_concurrent_updates
+        self.semaphore = BoundedSemaphore(self.max_concurrent_updates or 1)
+
+    @abstractmethod
+    async def process_update(
+        self,
+        update: object,
+        coroutine: "Awaitable[Any]",
+    ) -> None:
+        pass
+
+    @abstractmethod
+    async def initialize(self) -> None:
+        pass
+
+    @abstractmethod
+    async def shutdown(self) -> None:
+        pass
+
+    async def do_process_update(
+        self,
+        update: object,
+        coroutine: "Awaitable[Any]",
+    ) -> None:
+        async with self.semaphore:
+            await self.process_update(update, coroutine)
+
+    async def __aenter__(self) -> "BaseProcessor":
+        try:
+            await self.initialize()
+            return self
+        except Exception as exc:
+            await self.shutdown()
+            raise exc
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> None:
+        await self.shutdown()
+
+
+class SimpleProcessor(BaseProcessor):
+    async def process_update(
+        self,
+        update: object,
+        coroutine: "Awaitable[Any]",
+    ) -> None:
+        await coroutine
+
+    async def initialize(self) -> None:
+        pass
+
+    async def shutdown(self) -> None:
+        pass

telegram/ext/_applicationbuilder.py

@@ -35,7 +35,7 @@
 from telegram._bot import Bot
 from telegram._utils.defaultvalue import DEFAULT_FALSE, DEFAULT_NONE, DefaultValue
 from telegram._utils.types import DVInput, DVType, FilePathInput, ODVInput
-from telegram.ext._application import Application
+from telegram.ext._application import Application, BaseProcessor
 from telegram.ext._contexttypes import ContextTypes
 from telegram.ext._extbot import ExtBot
 from telegram.ext._jobqueue import JobQueue
@@ -198,7 +198,7 @@ def __init__(self: "InitApplicationBuilder"):
         self._context_types: DVType[ContextTypes] = DefaultValue(ContextTypes())
         self._application_class: DVType[Type[Application]] = DefaultValue(Application)
         self._application_kwargs: Dict[str, object] = {}
-        self._concurrent_updates: Union[int, DefaultValue[bool]] = DEFAULT_FALSE
+        self._concurrent_updates: Union[int, DefaultValue[bool], "BaseProcessor"] = DEFAULT_FALSE
         self._updater: ODVInput[Updater] = DEFAULT_NONE
         self._post_init: Optional[Callable[[Application], Coroutine[Any, Any, None]]] = None
         self._post_shutdown: Optional[Callable[[Application], Coroutine[Any, Any, None]]] = None
@@ -901,7 +901,9 @@ def update_queue(self: BuilderType, update_queue: "Queue[object]") -> BuilderTyp
         self._update_queue = update_queue
         return self
 
-    def concurrent_updates(self: BuilderType, concurrent_updates: Union[bool, int]) -> BuilderType:
+    def concurrent_updates(
+        self: BuilderType, concurrent_updates: Union[bool, int, "BaseProcessor"]
+    ) -> BuilderType:
         """Specifies if and how many updates may be processed concurrently instead of one by one.
         If not called, updates will be processed one by one.
 
@@ -916,9 +918,11 @@ def concurrent_updates(self: BuilderType, concurrent_updates: Union[bool, int])
         .. seealso:: :attr:`telegram.ext.Application.concurrent_updates`
 
         Args:
-            concurrent_updates (:obj:`bool` | :obj:`int`): Passing :obj:`True` will allow for
-                ``256`` updates to be processed concurrently. Pass an integer to specify a
-                different number of updates that may be processed concurrently.
+            concurrent_updates (:obj:`bool` | :obj:`int` | :class:`BaseProcessor`): Passing
+            :obj:`True` will allow for `256`` updates to be processed concurrently. Pass an
+            integer to specify a different number of updates that may be processed
+            concurrently. Pass an instance of `BaseProcessor` to use that instance
+            for handling updates concurrently.
 
         Returns:
             :class:`ApplicationBuilder`: The same builder with the updated argument.