#!/usr/bin/env python # # A library that provides a Python interface to the Telegram Bot API # Copyright (C) 2015-2022 # Leandro Toledo de Souza # # 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 base class for handlers as used by the Application.""" from abc import ABC, abstractmethod from typing import TYPE_CHECKING, Any, Generic, Optional, TypeVar, Union from telegram._utils.defaultvalue import DEFAULT_TRUE from telegram._utils.types import DVInput from telegram.ext._utils.types import CCT, HandlerCallback if TYPE_CHECKING: from telegram.ext import Application RT = TypeVar('RT') UT = TypeVar('UT') class Handler(Generic[UT, CCT], ABC): """The base class for all update handlers. Create custom handlers by inheriting from it. Warning: When setting :paramref:`block` to :obj:`False`, you cannot rely on adding custom attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info. .. versionchanged:: 14.0 The attribute ``run_async`` is now :paramref:`block`. Args: callback (:term:`coroutine function`): The callback function for this handler. Will be called when :meth:`check_update` has determined that an update should be processed by this handler. Callback signature:: async def callback(update: Update, context: CallbackContext) The return value of the callback is usually ignored except for the special case of :class:`telegram.ext.ConversationHandler`. block (:obj:`bool`, optional): Determines whether the return value of the callback should be awaited before processing the next handler in :meth:`telegram.ext.Application.process_update`. Defaults to :obj:`True`. Attributes: callback (:term:`coroutine function`): The callback function for this handler. block (:obj:`bool`): Determines whether the callback will run in a blocking way.. """ __slots__ = ( 'callback', 'block', ) def __init__( self, callback: HandlerCallback[UT, CCT, RT], block: DVInput[bool] = DEFAULT_TRUE, ): self.callback = callback self.block = block @abstractmethod def check_update(self, update: object) -> Optional[Union[bool, object]]: """ This method is called to determine if an update should be handled by this handler instance. It should always be overridden. Note: Custom updates types can be handled by the application. Therefore, an implementation of this method should always check the type of :paramref:`update`. Args: update (:obj:`object` | :class:`telegram.Update`): The update to be tested. Returns: Either :obj:`None` or :obj:`False` if the update should not be handled. Otherwise an object that will be passed to :meth:`handle_update` and :meth:`collect_additional_context` when the update gets handled. """ async def handle_update( self, update: UT, application: 'Application', check_result: object, context: CCT, ) -> RT: """ This method is called if it was determined that an update should indeed be handled by this instance. Calls :attr:`callback` along with its respectful arguments. To work with the :class:`telegram.ext.ConversationHandler`, this method returns the value returned from :attr:`callback`. Note that it can be overridden if needed by the subclassing handler. Args: update (:obj:`str` | :class:`telegram.Update`): The update to be handled. application (:class:`telegram.ext.Application`): The calling application. check_result (:class:`object`): The result from :meth:`check_update`. context (:class:`telegram.ext.CallbackContext`): The context as provided by the application. """ self.collect_additional_context(context, update, application, check_result) return await self.callback(update, context) def collect_additional_context( self, context: CCT, update: UT, application: 'Application', check_result: Any, ) -> None: """Prepares additional arguments for the context. Override if needed. Args: context (:class:`telegram.ext.CallbackContext`): The context object. update (:class:`telegram.Update`): The update to gather chat/user id from. application (:class:`telegram.ext.Application`): The calling application. check_result: The result (return value) from :meth:`check_update`. """