Improve Annotations & Docs of Handlers (#2243)

* Improve typing & docs of handlers

* Apply suggestions from code review

Co-authored-by: Poolitzer <25934244+Poolitzer@users.noreply.github.com>

Co-authored-by: Poolitzer <25934244+Poolitzer@users.noreply.github.com>
This commit is contained in:
Bibo-Joshi 2020-12-16 17:34:57 +01:00 committed by GitHub
parent 786762bb73
commit 77a8c64f6c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 103 additions and 107 deletions

View file

@ -33,7 +33,6 @@ from typing import (
)
from telegram import Update
from telegram.utils.types import HandlerArg
from telegram.utils.helpers import DefaultValue, DEFAULT_FALSE
from .handler import Handler
@ -44,7 +43,7 @@ if TYPE_CHECKING:
RT = TypeVar('RT')
class CallbackQueryHandler(Handler):
class CallbackQueryHandler(Handler[Update]):
"""Handler class to handle Telegram callback queries. Optionally based on a regex.
Read the documentation of the ``re`` module for more information.
@ -123,7 +122,7 @@ class CallbackQueryHandler(Handler):
def __init__(
self,
callback: Callable[[HandlerArg, 'CallbackContext'], RT],
callback: Callable[[Update, 'CallbackContext'], RT],
pass_update_queue: bool = False,
pass_job_queue: bool = False,
pattern: Union[str, Pattern] = None,
@ -149,11 +148,11 @@ class CallbackQueryHandler(Handler):
self.pass_groups = pass_groups
self.pass_groupdict = pass_groupdict
def check_update(self, update: HandlerArg) -> Optional[Union[bool, object]]:
def check_update(self, update: Any) -> Optional[Union[bool, object]]:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`bool`
@ -172,7 +171,7 @@ class CallbackQueryHandler(Handler):
def collect_optional_args(
self,
dispatcher: 'Dispatcher',
update: HandlerArg = None,
update: Update = None,
check_result: Union[bool, Match] = None,
) -> Dict[str, Any]:
optional_args = super().collect_optional_args(dispatcher, update, check_result)
@ -187,7 +186,7 @@ class CallbackQueryHandler(Handler):
def collect_additional_context(
self,
context: 'CallbackContext',
update: HandlerArg,
update: Update,
dispatcher: 'Dispatcher',
check_result: Union[bool, Match],
) -> None:

View file

@ -18,17 +18,16 @@
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""This module contains the ChosenInlineResultHandler class."""
from typing import Optional, TypeVar, Union
from typing import Optional, TypeVar, Union, Any
from telegram import Update
from telegram.utils.types import HandlerArg
from .handler import Handler
RT = TypeVar('RT')
class ChosenInlineResultHandler(Handler):
class ChosenInlineResultHandler(Handler[Update]):
"""Handler class to handle Telegram updates that contain a chosen inline result.
Attributes:
@ -86,11 +85,11 @@ class ChosenInlineResultHandler(Handler):
"""
def check_update(self, update: HandlerArg) -> Optional[Union[bool, object]]:
def check_update(self, update: Any) -> Optional[Union[bool, object]]:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`bool`

View file

@ -24,7 +24,7 @@ from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Ty
from telegram import MessageEntity, Update
from telegram.ext import BaseFilter, Filters
from telegram.utils.deprecate import TelegramDeprecationWarning
from telegram.utils.types import HandlerArg, SLT
from telegram.utils.types import SLT
from telegram.utils.helpers import DefaultValue, DEFAULT_FALSE
from .handler import Handler
@ -35,7 +35,7 @@ if TYPE_CHECKING:
RT = TypeVar('RT')
class CommandHandler(Handler):
class CommandHandler(Handler[Update]):
"""Handler class to handle Telegram commands.
Commands are Telegram messages that start with ``/``, optionally followed by an ``@`` and the
@ -134,7 +134,7 @@ class CommandHandler(Handler):
def __init__(
self,
command: SLT[str],
callback: Callable[[HandlerArg, 'CallbackContext'], RT],
callback: Callable[[Update, 'CallbackContext'], RT],
filters: BaseFilter = None,
allow_edited: bool = None,
pass_args: bool = False,
@ -177,12 +177,12 @@ class CommandHandler(Handler):
self.pass_args = pass_args
def check_update(
self, update: HandlerArg
self, update: Any
) -> Optional[Union[bool, Tuple[List[str], Optional[Union[bool, Dict]]]]]:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`list`: The list of args for the handler.
@ -218,7 +218,7 @@ class CommandHandler(Handler):
def collect_optional_args(
self,
dispatcher: 'Dispatcher',
update: HandlerArg = None,
update: Update = None,
check_result: Optional[Union[bool, Tuple[List[str], Optional[bool]]]] = None,
) -> Dict[str, Any]:
optional_args = super().collect_optional_args(dispatcher, update)
@ -229,7 +229,7 @@ class CommandHandler(Handler):
def collect_additional_context(
self,
context: 'CallbackContext',
update: HandlerArg,
update: Update,
dispatcher: 'Dispatcher',
check_result: Optional[Union[bool, Tuple[List[str], Optional[bool]]]],
) -> None:
@ -344,7 +344,7 @@ class PrefixHandler(CommandHandler):
self,
prefix: SLT[str],
command: SLT[str],
callback: Callable[[HandlerArg, 'CallbackContext'], RT],
callback: Callable[[Update, 'CallbackContext'], RT],
filters: BaseFilter = None,
pass_args: bool = False,
pass_update_queue: bool = False,
@ -415,12 +415,12 @@ class PrefixHandler(CommandHandler):
self._commands = [x.lower() + y.lower() for x in self.prefix for y in self.command]
def check_update(
self, update: HandlerArg
self, update: Update
) -> Optional[Union[bool, Tuple[List[str], Optional[Union[bool, Dict]]]]]:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`list`: The list of args for the handler.
@ -442,7 +442,7 @@ class PrefixHandler(CommandHandler):
def collect_additional_context(
self,
context: 'CallbackContext',
update: HandlerArg,
update: Update,
dispatcher: 'Dispatcher',
check_result: Optional[Union[bool, Tuple[List[str], Optional[bool]]]],
) -> None:

View file

@ -35,7 +35,7 @@ from telegram.ext import (
InlineQueryHandler,
)
from telegram.utils.promise import Promise
from telegram.utils.types import ConversationDict, HandlerArg
from telegram.utils.types import ConversationDict
if TYPE_CHECKING:
from telegram.ext import Dispatcher, Job
@ -56,7 +56,7 @@ class _ConversationTimeoutContext:
self.callback_context = callback_context
class ConversationHandler(Handler):
class ConversationHandler(Handler[Update]):
"""
A handler to hold a conversation with a single user by managing four collections of other
handlers.
@ -391,13 +391,13 @@ class ConversationHandler(Handler):
return tuple(key)
def check_update(self, update: HandlerArg) -> CheckUpdateType: # pylint: disable=R0911
def check_update(self, update: Any) -> CheckUpdateType: # pylint: disable=R0911
"""
Determines whether an update should be handled by this conversationhandler, and if so in
which state the conversation currently is.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`bool`
@ -487,7 +487,7 @@ class ConversationHandler(Handler):
def handle_update( # type: ignore[override]
self,
update: HandlerArg,
update: Update,
dispatcher: 'Dispatcher',
check_result: CheckUpdateType,
context: CallbackContext = None,

View file

@ -35,7 +35,6 @@ from telegram.ext.callbackcontext import CallbackContext
from telegram.ext.handler import Handler
from telegram.utils.deprecate import TelegramDeprecationWarning
from telegram.utils.promise import Promise
from telegram.utils.types import HandlerArg
from telegram.utils.helpers import DefaultValue, DEFAULT_FALSE
if TYPE_CHECKING:
@ -288,7 +287,7 @@ class Dispatcher:
self.logger.exception('An uncaught error was raised while handling the error.')
def run_async(
self, func: Callable[..., Any], *args: Any, update: HandlerArg = None, **kwargs: Any
self, func: Callable[..., Any], *args: Any, update: Any = None, **kwargs: Any
) -> Promise:
"""
Queue a function (with given args/kwargs) to be run asynchronously. Exceptions raised
@ -304,9 +303,9 @@ class Dispatcher:
Args:
func (:obj:`callable`): The function to run in the thread.
*args (:obj:`tuple`, optional): Arguments to ``func``.
update (:class:`telegram.Update`, optional): The update associated with the functions
call. If passed, it will be available in the error handlers, in case an exception
is raised by :attr:`func`.
update (:class:`telegram.Update` | :obj:`object`, optional): The update associated with
the functions call. If passed, it will be available in the error handlers, in case
an exception is raised by :attr:`func`.
**kwargs (:obj:`dict`, optional): Keyword arguments to ``func``.
Returns:
@ -319,7 +318,7 @@ class Dispatcher:
self,
func: Callable[..., Any],
*args: Any,
update: HandlerArg = None,
update: Any = None,
error_handling: bool = True,
**kwargs: Any,
) -> Promise:
@ -515,7 +514,7 @@ class Dispatcher:
del self.handlers[group]
self.groups.remove(group)
def update_persistence(self, update: HandlerArg = None) -> None:
def update_persistence(self, update: Any = None) -> None:
"""Update :attr:`user_data`, :attr:`chat_data` and :attr:`bot_data` in :attr:`persistence`.
Args:
@ -525,7 +524,7 @@ class Dispatcher:
with self._update_persistence_lock:
self.__update_persistence(update)
def __update_persistence(self, update: HandlerArg = None) -> None:
def __update_persistence(self, update: Any = None) -> None:
if self.persistence:
# We use list() here in order to decouple chat_ids from self.chat_data, as dict view
# objects will change, when the dict does and we want to loop over chat_ids
@ -632,12 +631,12 @@ class Dispatcher:
self.error_handlers.pop(callback, None)
def dispatch_error(
self, update: Optional[HandlerArg], error: Exception, promise: Promise = None
self, update: Optional[Any], error: Exception, promise: Promise = None
) -> None:
"""Dispatches an error.
Args:
update (:obj:`str` | :class:`telegram.Update` | None): The update that caused the error
update (:obj:`Any` | :class:`telegram.Update` | None): The update that caused the error.
error (:obj:`Exception`): The error that was raised.
promise (:class:`telegram.utils.Promise`, optional): The promise whose pooled function
raised the error.

View file

@ -19,20 +19,20 @@
"""This module contains the base class for handlers as used by the Dispatcher."""
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, TypeVar, Union
from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, TypeVar, Union, Generic
from telegram import Update
from telegram.utils.promise import Promise
from telegram.utils.types import HandlerArg
from telegram.utils.helpers import DefaultValue, DEFAULT_FALSE
if TYPE_CHECKING:
from telegram.ext import CallbackContext, Dispatcher
RT = TypeVar('RT')
UT = TypeVar('UT')
class Handler(ABC):
class Handler(Generic[UT], ABC):
"""The base class for all update handlers. Create custom handlers by inheriting from it.
Attributes:
@ -92,14 +92,14 @@ class Handler(ABC):
def __init__(
self,
callback: Callable[[HandlerArg, 'CallbackContext'], RT],
callback: Callable[[UT, 'CallbackContext'], RT],
pass_update_queue: bool = False,
pass_job_queue: bool = False,
pass_user_data: bool = False,
pass_chat_data: bool = False,
run_async: Union[bool, DefaultValue] = DEFAULT_FALSE,
):
self.callback: Callable[[HandlerArg, 'CallbackContext'], RT] = callback
self.callback = callback
self.pass_update_queue = pass_update_queue
self.pass_job_queue = pass_job_queue
self.pass_user_data = pass_user_data
@ -107,11 +107,15 @@ class Handler(ABC):
self.run_async = run_async
@abstractmethod
def check_update(self, update: HandlerArg) -> Optional[Union[bool, object]]:
def check_update(self, update: Any) -> 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 dispatcher. Therefore, an implementation of
this method should always check the type of :attr:`update`.
Args:
update (:obj:`str` | :class:`telegram.Update`): The update to be tested.
@ -124,7 +128,7 @@ class Handler(ABC):
def handle_update(
self,
update: HandlerArg,
update: UT,
dispatcher: 'Dispatcher',
check_result: object,
context: 'CallbackContext' = None,
@ -165,7 +169,7 @@ class Handler(ABC):
def collect_additional_context(
self,
context: 'CallbackContext',
update: HandlerArg,
update: UT,
dispatcher: 'Dispatcher',
check_result: Any,
) -> None:
@ -182,7 +186,7 @@ class Handler(ABC):
def collect_optional_args(
self,
dispatcher: 'Dispatcher',
update: HandlerArg = None,
update: UT = None,
check_result: Any = None, # pylint: disable=W0613
) -> Dict[str, Any]:
"""

View file

@ -32,7 +32,6 @@ from typing import (
)
from telegram import Update
from telegram.utils.types import HandlerArg
from telegram.utils.helpers import DefaultValue, DEFAULT_FALSE
from .handler import Handler
@ -43,7 +42,7 @@ if TYPE_CHECKING:
RT = TypeVar('RT')
class InlineQueryHandler(Handler):
class InlineQueryHandler(Handler[Update]):
"""
Handler class to handle Telegram inline queries. Optionally based on a regex. Read the
documentation of the ``re`` module for more information.
@ -122,7 +121,7 @@ class InlineQueryHandler(Handler):
def __init__(
self,
callback: Callable[[HandlerArg, 'CallbackContext'], RT],
callback: Callable[[Update, 'CallbackContext'], RT],
pass_update_queue: bool = False,
pass_job_queue: bool = False,
pattern: Union[str, Pattern] = None,
@ -148,12 +147,12 @@ class InlineQueryHandler(Handler):
self.pass_groups = pass_groups
self.pass_groupdict = pass_groupdict
def check_update(self, update: HandlerArg) -> Optional[Union[bool, Match]]:
def check_update(self, update: Any) -> Optional[Union[bool, Match]]:
"""
Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`bool`
@ -173,7 +172,7 @@ class InlineQueryHandler(Handler):
def collect_optional_args(
self,
dispatcher: 'Dispatcher',
update: HandlerArg = None,
update: Update = None,
check_result: Optional[Union[bool, Match]] = None,
) -> Dict[str, Any]:
optional_args = super().collect_optional_args(dispatcher, update, check_result)
@ -188,7 +187,7 @@ class InlineQueryHandler(Handler):
def collect_additional_context(
self,
context: 'CallbackContext',
update: HandlerArg,
update: Update,
dispatcher: 'Dispatcher',
check_result: Optional[Union[bool, Match]],
) -> None:

View file

@ -24,7 +24,6 @@ from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, TypeVar, Union
from telegram import Update
from telegram.ext import BaseFilter, Filters
from telegram.utils.deprecate import TelegramDeprecationWarning
from telegram.utils.types import HandlerArg
from telegram.utils.helpers import DefaultValue, DEFAULT_FALSE
from .handler import Handler
@ -35,7 +34,7 @@ if TYPE_CHECKING:
RT = TypeVar('RT')
class MessageHandler(Handler):
class MessageHandler(Handler[Update]):
"""Handler class to handle telegram messages. They might contain text, media or status updates.
Attributes:
@ -124,7 +123,7 @@ class MessageHandler(Handler):
def __init__(
self,
filters: BaseFilter,
callback: Callable[[HandlerArg, 'CallbackContext'], RT],
callback: Callable[[Update, 'CallbackContext'], RT],
pass_update_queue: bool = False,
pass_job_queue: bool = False,
pass_user_data: bool = False,
@ -180,11 +179,11 @@ class MessageHandler(Handler):
Filters.update.edited_message | Filters.update.edited_channel_post
)
def check_update(self, update: HandlerArg) -> Optional[Union[bool, Dict[str, Any]]]:
def check_update(self, update: Any) -> Optional[Union[bool, Dict[str, Any]]]:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`bool`
@ -197,7 +196,7 @@ class MessageHandler(Handler):
def collect_additional_context(
self,
context: 'CallbackContext',
update: HandlerArg,
update: Update,
dispatcher: 'Dispatcher',
check_result: Optional[Union[bool, Dict[str, Any]]],
) -> None:

View file

@ -17,14 +17,14 @@
# 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 PollAnswerHandler class."""
from typing import Any
from telegram import Update
from telegram.utils.types import HandlerArg
from .handler import Handler
class PollAnswerHandler(Handler):
class PollAnswerHandler(Handler[Update]):
"""Handler class to handle Telegram updates that contain a poll answer.
Attributes:
@ -82,11 +82,11 @@ class PollAnswerHandler(Handler):
"""
def check_update(self, update: HandlerArg) -> bool:
def check_update(self, update: Any) -> bool:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`bool`

View file

@ -17,14 +17,14 @@
# 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 PollHandler classes."""
from typing import Any
from telegram import Update
from telegram.utils.types import HandlerArg
from .handler import Handler
class PollHandler(Handler):
class PollHandler(Handler[Update]):
"""Handler class to handle Telegram updates that contain a poll.
Attributes:
@ -82,11 +82,11 @@ class PollHandler(Handler):
"""
def check_update(self, update: HandlerArg) -> bool:
def check_update(self, update: Any) -> bool:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`bool`

View file

@ -17,14 +17,14 @@
# 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 PreCheckoutQueryHandler class."""
from typing import Any
from telegram import Update
from telegram.utils.types import HandlerArg
from .handler import Handler
class PreCheckoutQueryHandler(Handler):
class PreCheckoutQueryHandler(Handler[Update]):
"""Handler class to handle Telegram PreCheckout callback queries.
Attributes:
@ -82,11 +82,11 @@ class PreCheckoutQueryHandler(Handler):
"""
def check_update(self, update: HandlerArg) -> bool:
def check_update(self, update: Any) -> bool:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`bool`

View file

@ -22,9 +22,9 @@
import warnings
from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Pattern, TypeVar, Union
from telegram import Update
from telegram.ext import Filters, MessageHandler
from telegram.utils.deprecate import TelegramDeprecationWarning
from telegram.utils.types import HandlerArg
from telegram.utils.helpers import DefaultValue, DEFAULT_FALSE
if TYPE_CHECKING:
@ -111,7 +111,7 @@ class RegexHandler(MessageHandler):
def __init__(
self,
pattern: Union[str, Pattern],
callback: Callable[[HandlerArg, 'CallbackContext'], RT],
callback: Callable[[Update, 'CallbackContext'], RT],
pass_groups: bool = False,
pass_groupdict: bool = False,
pass_update_queue: bool = False,
@ -147,7 +147,7 @@ class RegexHandler(MessageHandler):
def collect_optional_args(
self,
dispatcher: 'Dispatcher',
update: HandlerArg = None,
update: Update = None,
check_result: Optional[Union[bool, Dict[str, Any]]] = None,
) -> Dict[str, Any]:
optional_args = super().collect_optional_args(dispatcher, update, check_result)

View file

@ -17,14 +17,13 @@
# 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 ShippingQueryHandler class."""
from typing import Any
from telegram import Update
from telegram.utils.types import HandlerArg
from .handler import Handler
class ShippingQueryHandler(Handler):
class ShippingQueryHandler(Handler[Update]):
"""Handler class to handle Telegram shipping callback queries.
Attributes:
@ -82,11 +81,11 @@ class ShippingQueryHandler(Handler):
"""
def check_update(self, update: HandlerArg) -> bool:
def check_update(self, update: Any) -> bool:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`bool`

View file

@ -20,7 +20,6 @@
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, TypeVar, Union
from telegram.utils.types import HandlerArg
from telegram.utils.helpers import DefaultValue, DEFAULT_FALSE
from .handler import Handler
@ -31,7 +30,7 @@ if TYPE_CHECKING:
RT = TypeVar('RT')
class StringCommandHandler(Handler):
class StringCommandHandler(Handler[str]):
"""Handler class to handle string commands. Commands are string updates that start with ``/``.
Note:
@ -86,7 +85,7 @@ class StringCommandHandler(Handler):
def __init__(
self,
command: str,
callback: Callable[[HandlerArg, 'CallbackContext'], RT],
callback: Callable[[str, 'CallbackContext'], RT],
pass_args: bool = False,
pass_update_queue: bool = False,
pass_job_queue: bool = False,
@ -101,11 +100,11 @@ class StringCommandHandler(Handler):
self.command = command
self.pass_args = pass_args
def check_update(self, update: HandlerArg) -> Optional[List[str]]:
def check_update(self, update: Any) -> Optional[List[str]]:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:obj:`str`): An incoming command.
update (:obj:`object`): The incoming update.
Returns:
:obj:`bool`
@ -120,7 +119,7 @@ class StringCommandHandler(Handler):
def collect_optional_args(
self,
dispatcher: 'Dispatcher',
update: HandlerArg = None,
update: str = None,
check_result: Optional[List[str]] = None,
) -> Dict[str, Any]:
optional_args = super().collect_optional_args(dispatcher, update, check_result)
@ -131,7 +130,7 @@ class StringCommandHandler(Handler):
def collect_additional_context(
self,
context: 'CallbackContext',
update: HandlerArg,
update: str,
dispatcher: 'Dispatcher',
check_result: Optional[List[str]],
) -> None:

View file

@ -21,7 +21,6 @@
import re
from typing import TYPE_CHECKING, Any, Callable, Dict, Match, Optional, Pattern, TypeVar, Union
from telegram.utils.types import HandlerArg
from telegram.utils.helpers import DefaultValue, DEFAULT_FALSE
from .handler import Handler
@ -32,7 +31,7 @@ if TYPE_CHECKING:
RT = TypeVar('RT')
class StringRegexHandler(Handler):
class StringRegexHandler(Handler[str]):
"""Handler class to handle string updates based on a regex which checks the update content.
Read the documentation of the ``re`` module for more information. The ``re.match`` function is
@ -95,7 +94,7 @@ class StringRegexHandler(Handler):
def __init__(
self,
pattern: Union[str, Pattern],
callback: Callable[[HandlerArg, 'CallbackContext'], RT],
callback: Callable[[str, 'CallbackContext'], RT],
pass_groups: bool = False,
pass_groupdict: bool = False,
pass_update_queue: bool = False,
@ -116,11 +115,11 @@ class StringRegexHandler(Handler):
self.pass_groups = pass_groups
self.pass_groupdict = pass_groupdict
def check_update(self, update: HandlerArg) -> Optional[Match]:
def check_update(self, update: Any) -> Optional[Match]:
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:obj:`str`): An incoming command.
update (:obj:`object`): The incoming update.
Returns:
:obj:`bool`
@ -135,7 +134,7 @@ class StringRegexHandler(Handler):
def collect_optional_args(
self,
dispatcher: 'Dispatcher',
update: HandlerArg = None,
update: str = None,
check_result: Optional[Match] = None,
) -> Dict[str, Any]:
optional_args = super().collect_optional_args(dispatcher, update, check_result)
@ -149,7 +148,7 @@ class StringRegexHandler(Handler):
def collect_additional_context(
self,
context: 'CallbackContext',
update: HandlerArg,
update: str,
dispatcher: 'Dispatcher',
check_result: Optional[Match],
) -> None:

View file

@ -27,9 +27,10 @@ if TYPE_CHECKING:
from telegram.ext import CallbackContext
RT = TypeVar('RT')
UT = TypeVar('UT')
class TypeHandler(Handler):
class TypeHandler(Handler[UT]):
"""Handler class to handle updates of custom types.
Attributes:
@ -76,8 +77,8 @@ class TypeHandler(Handler):
def __init__(
self,
type: Type, # pylint: disable=W0622
callback: Callable[[Any, 'CallbackContext'], RT],
type: Type[UT], # pylint: disable=W0622
callback: Callable[[UT, 'CallbackContext'], RT],
strict: bool = False,
pass_update_queue: bool = False,
pass_job_queue: bool = False,
@ -96,7 +97,7 @@ class TypeHandler(Handler):
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
update (:obj:`object`): Incoming update.
Returns:
:obj:`bool`

View file

@ -20,9 +20,9 @@
import logging
from threading import Event
from typing import Callable, List, Optional, Tuple, TypeVar, Union
from typing import Callable, List, Optional, Tuple, TypeVar, Union, Any
from telegram.utils.types import HandlerArg, JSONDict
from telegram.utils.types import JSONDict
RT = TypeVar('RT')
@ -37,7 +37,8 @@ class Promise:
pooled_function (:obj:`callable`): The callable that will be called concurrently.
args (:obj:`list` | :obj:`tuple`): Positional arguments for :attr:`pooled_function`.
kwargs (:obj:`dict`): Keyword arguments for :attr:`pooled_function`.
update (:class:`telegram.Update`, optional): The update this promise is associated with.
update (:class:`telegram.Update` | :obj:`object`, optional): The update this promise is
associated with.
error_handling (:obj:`bool`, optional): Whether exceptions raised by :attr:`func`
may be handled by error handlers. Defaults to :obj:`True`.
@ -46,7 +47,8 @@ class Promise:
args (:obj:`list` | :obj:`tuple`): Positional arguments for :attr:`pooled_function`.
kwargs (:obj:`dict`): Keyword arguments for :attr:`pooled_function`.
done (:obj:`threading.Event`): Is set when the result is available.
update (:class:`telegram.Update`): Optional. The update this promise is associated with.
update (:class:`telegram.Update` | :obj:`object`): Optional. The update this promise is
associated with.
error_handling (:obj:`bool`): Optional. Whether exceptions raised by :attr:`func`
may be handled by error handlers. Defaults to :obj:`True`.
@ -58,7 +60,7 @@ class Promise:
pooled_function: Callable[..., RT],
args: Union[List, Tuple],
kwargs: JSONDict,
update: HandlerArg = None,
update: Any = None,
error_handling: bool = True,
):
self.pooled_function = pooled_function

View file

@ -21,7 +21,7 @@ from pathlib import Path
from typing import IO, TYPE_CHECKING, Any, Dict, List, Optional, Tuple, TypeVar, Union
if TYPE_CHECKING:
from telegram import InputFile, Update
from telegram import InputFile
FileLike = Union[IO, 'InputFile']
"""Either an open file handler or a :class:`telegram.InputFile`."""
@ -33,9 +33,6 @@ a local file path as string or :class:`pathlib.Path`."""
JSONDict = Dict[str, Any]
"""Dictionary containing response from Telegram or data to send to the API."""
HandlerArg = Union[str, 'Update']
"""The argument that handlers parse for :meth:`telegram.ext.handler.check_update` etc."""
ConversationDict = Dict[Tuple[int, ...], Optional[object]]
"""Dicts as maintained by the :class:`telegram.ext.ConversationHandler`."""