Customized concurrent handling

docs/source/telegram.ext.baseupdateprocessor.rst

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

docs/source/telegram.ext.simpleupdateprocessor.rst

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

telegram/ext/__init__.py

@@ -25,6 +25,7 @@
     "ApplicationHandlerStop",
     "BaseHandler",
     "BasePersistence",
+    "BaseUpdateProcessor",
     "BaseRateLimiter",
     "CallbackContext",
     "CallbackDataCache",
@@ -51,6 +52,7 @@
     "PreCheckoutQueryHandler",
     "PrefixHandler",
     "ShippingQueryHandler",
+    "SimpleUpdateProcessor",
     "StringCommandHandler",
     "StringRegexHandler",
     "TypeHandler",
@@ -63,6 +65,7 @@
 from ._applicationbuilder import ApplicationBuilder
 from ._basepersistence import BasePersistence, PersistenceInput
 from ._baseratelimiter import BaseRateLimiter
+from ._baseupdateprocessor import BaseUpdateProcessor, SimpleUpdateProcessor
 from ._callbackcontext import CallbackContext
 from ._callbackdatacache import CallbackDataCache, InvalidCallbackData
 from ._callbackqueryhandler import CallbackQueryHandler

telegram/ext/_application.py

@@ -57,6 +57,7 @@
 from telegram._utils.warnings import warn
 from telegram.error import TelegramError
 from telegram.ext._basepersistence import BasePersistence
+from telegram.ext._baseupdateprocessor import BaseUpdateProcessor
 from telegram.ext._contexttypes import ContextTypes
 from telegram.ext._extbot import ExtBot
 from telegram.ext._handler import BaseHandler
@@ -259,7 +260,7 @@ def __init__(
         update_queue: "asyncio.Queue[object]",
         updater: Optional[Updater],
         job_queue: JQ,
-        concurrent_updates: Union[bool, int],
+        update_processor: "BaseUpdateProcessor",
         persistence: Optional[BasePersistence[UD, CD, BD]],
         context_types: ContextTypes[CCT, UD, CD, BD],
         post_init: Optional[
@@ -297,14 +298,10 @@ 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
-
+        self._concurrent_updates_sem = asyncio.BoundedSemaphore(
+            update_processor.max_concurrent_updates or 1
+        )
+        self._update_processor = update_processor
         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)
@@ -359,9 +356,12 @@ def concurrent_updates(self) -> int:
         """:obj:`int`: The number of concurrent updates that will be processed in parallel. A
         value of ``0`` indicates updates are *not* being processed concurrently.
 
+        Note:
+            This is now just a shortcut to `processor.max_concurrent_updates`.
+
         .. seealso:: :wiki:`Concurrency`
         """
-        return self._concurrent_updates
+        return self._update_processor.max_concurrent_updates
 
     @property
     def job_queue(self) -> Optional["JobQueue[CCT]"]:
@@ -379,6 +379,14 @@ def job_queue(self) -> Optional["JobQueue[CCT]"]:
             )
         return self._job_queue
 
+    @property
+    def processor(self) -> Optional["BaseUpdateProcessor"]:
+        """:class:`telegram.ext.BaseProcessor`: The processor used by this application.
+
+        .. seealso:: :wiki:`Concurrency`
+        """
+        return self._update_processor
+
     async def initialize(self) -> None:
         """Initializes the Application by initializing:
 
@@ -1059,9 +1067,14 @@ async def _update_fetcher(self) -> None:
 
             _LOGGER.debug("Processing update %s", update)
 
-            if self._concurrent_updates:
+            if self._update_processor and self._update_processor.max_concurrent_updates > 0:
                 # 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._update_processor.do_process_update(
+                        update, self.process_update(update), self
+                    ),
+                    update=update,
+                )
             else:
                 await self.__process_update_wrapper(update)
 

telegram/ext/_applicationbuilder.py

@@ -36,6 +36,7 @@
 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._baseupdateprocessor import BaseUpdateProcessor, SimpleUpdateProcessor
 from telegram.ext._contexttypes import ContextTypes
 from telegram.ext._extbot import ExtBot
 from telegram.ext._jobqueue import JobQueue
@@ -198,7 +199,9 @@ 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], "BaseUpdateProcessor"
+        ] = 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
@@ -306,7 +309,7 @@ def build(
             bot=bot,
             update_queue=update_queue,
             updater=updater,
-            concurrent_updates=DefaultValue.get_value(self._concurrent_updates),
+            update_processor=DefaultValue.get_value(self._concurrent_updates),
             job_queue=job_queue,
             persistence=persistence,
             context_types=DefaultValue.get_value(self._context_types),
@@ -901,7 +904,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, "BaseUpdateProcessor"]
+    ) -> 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,14 +921,33 @@ 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:`BaseUpdateProcessor`): 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,
+            in that case :class:`telegram.ext.SimpleUpdateProcessor` is used to handle updates.
+            Pass an instance of :class:`telegram.ext.BaseUpdateProcessor` to use that instance
+            for handling updates concurrently.
 
         Returns:
             :class:`ApplicationBuilder`: The same builder with the updated argument.
         """
-        self._concurrent_updates = concurrent_updates
+        # Check if concurrent updates is bool and convert to integer
+        if concurrent_updates is True:
+            concurrent_updates = 256
+        elif concurrent_updates is False:
+            concurrent_updates = 0
+
+        # If `concurrent_updates` is an integer, create a `SimpleUpdateProcessor`
+        # instance with that integer value; otherwise, raise an error if the value
+        # is negative
+        if isinstance(concurrent_updates, int):
+            if concurrent_updates < 0:
+                raise ValueError("`concurrent_updates` must be a non-negative integer!")
+            concurrent_updates = SimpleUpdateProcessor(concurrent_updates)
+
+        # Assign default value of concurrent_updates if it is instance of
+        # `BaseUpdateProcessor`
+        self._concurrent_updates: BaseUpdateProcessor = concurrent_updates
         return self
 
     def job_queue(

telegram/ext/_baseupdateprocessor.py

@@ -0,0 +1,140 @@
+#!/usr/bin/env python
+#
+# A library that provides a Python interface to the Telegram Bot API
+# Copyright (C) 2015-2023
+# Leandro Toledo de Souza <devs@python-telegram-bot.org>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program 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 Public License for more details.
+#
+# You should have received a copy of the GNU Lesser Public License
+# along with this program.  If not, see [http://www.gnu.org/licenses/].
+"""This module contains the BaseProcessor class."""
+from abc import abstractmethod
+from asyncio import BoundedSemaphore
+from types import TracebackType
+from typing import Any, Awaitable, Optional, Type
+
+from telegram.ext import Application
+
+
+class BaseUpdateProcessor:
+    """An abstract base class for update processors. You can use this class to implement
+    your own update processor.
+
+    .. seealso:: :wiki:`Concurrency`
+
+    .. versionadded:: NEXT.VERSION
+
+    Args:
+        max_concurrent_updates (:obj:`int`): The maximum number of concurrent updates to be
+            processed. If this number is exceeded, new updates will be queued until the number of
+            currently processed updates decreases.
+
+    Raises:
+        :exc:`ValueError`: If `max_concurrent_updates` is a negative integer.
+    """
+
+    def __init__(self, max_concurrent_updates: int):
+        self._max_concurrent_updates = max_concurrent_updates
+        if self.max_concurrent_updates < 0:
+            raise ValueError("`max_concurrent_updates` must be a non-negative integer!")
+        self._semaphore = BoundedSemaphore(self.max_concurrent_updates or 1)
+
+    @property
+    def max_concurrent_updates(self) -> int:
+        """:obj:`int`: The maximum number of updates that can be processed concurrently."""
+        return self._max_concurrent_updates
+
+    @abstractmethod
+    async def process_update(
+        self,
+        update: object,
+        coroutine: "Awaitable[Any]",
+    ) -> None:
+        """Custom implementation of how to process an update.
+
+        Args:
+            update (:obj:`object`): The update to be processed.
+            coroutine (:obj:`Awaitable`): The coroutine that will be awaited to process the update.
+        """
+
+    @abstractmethod
+    async def initialize(self) -> None:
+        """Initializes the Processor so resources can be allocated.
+
+        .. seealso::
+            :meth:`shutdown`
+        """
+
+    @abstractmethod
+    async def shutdown(self) -> None:
+        """Shutdown the Processor so resources can be freed.
+
+        .. seealso::
+            :meth:`initialize`
+        """
+
+    async def do_process_update(
+        self,
+        update: object,
+        coroutine: "Awaitable[Any]",
+        application: Application,
+    ) -> None:
+        """Calls :meth:`process_update` with a semaphore to limit the number of concurrent
+        updates.
+
+        Args:
+            update (:obj:`object`): The update to be processed.
+            coroutine (:obj:`Awaitable`): The coroutine that will be awaited to process the update.
+            application (:class:`telegram.ext.Application`): The application instance.
+        """
+        async with self._semaphore:
+            await self.process_update(update, coroutine)
+            application.update_queue.task_done()
+
+    async def __aenter__(self) -> "BaseUpdateProcessor":
+        """Simple context manager which initializes the Processor."""
+        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:
+        """Shutdown the Processor from the context manager."""
+        await self.shutdown()
+
+
+class SimpleUpdateProcessor(BaseUpdateProcessor):
+    async def process_update(
+        self,
+        update: object,
+        coroutine: "Awaitable[Any]",
+    ) -> None:
+        """Immediately awaits the coroutine, i.e. does not apply any additional processing.
+
+        Args:
+            update (:obj:`object`): The update to be processed.
+            coroutine (:obj:`Awaitable`): The coroutine that will be awaited to process the update.
+        """
+        await coroutine
+
+    async def initialize(self) -> None:
+        """Does nothing."""
+
+    async def shutdown(self) -> None:
+        """Does nothing."""