KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use run_task method in Lemoncheesecake

Best Python code snippet using lemoncheesecake

dispatcher.py

Source:dispatcher.py

1import asyncio2import functools3import itertools4import logging5import time6import typing7import aiohttp8from aiohttp.helpers import sentinel9from aiogram.utils.deprecated import renamed_argument10from .filters import Command, ContentTypeFilter, ExceptionsFilter, FiltersFactory, HashTag, Regexp, \11    RegexpCommandsFilter, StateFilter, Text, IDFilter, AdminFilter, IsReplyFilter12from .filters.builtin import IsSenderContact13from .handler import Handler14from .middlewares import MiddlewareManager15from .storage import BaseStorage, DELTA, DisabledStorage, EXCEEDED_COUNT, FSMContext, \16    LAST_CALL, RATE_LIMIT, RESULT17from .webhook import BaseResponse18from .. import types19from ..bot import Bot20from ..utils.exceptions import TelegramAPIError, Throttled21from ..utils.mixins import ContextInstanceMixin, DataMixin22log = logging.getLogger(__name__)23DEFAULT_RATE_LIMIT = .124class Dispatcher(DataMixin, ContextInstanceMixin):25    """26    Simple Updates dispatcher27    It will process incoming updates: messages, edited messages, channel posts, edited channel posts,28    inline queries, chosen inline results, callback queries, shipping queries, pre-checkout queries.29    """30    def __init__(self, bot, loop=None, storage: typing.Optional[BaseStorage] = None,31                 run_tasks_by_default: bool = False,32                 throttling_rate_limit=DEFAULT_RATE_LIMIT, no_throttle_error=False,33                 filters_factory=None):34        if not isinstance(bot, Bot):35            raise TypeError(f"Argument 'bot' must be an instance of Bot, not '{type(bot).__name__}'")36        if loop is None:37            loop = bot.loop38        if storage is None:39            storage = DisabledStorage()40        if filters_factory is None:41            filters_factory = FiltersFactory(self)42        self.bot: Bot = bot43        self.loop = loop44        self.storage = storage45        self.run_tasks_by_default = run_tasks_by_default46        self.throttling_rate_limit = throttling_rate_limit47        self.no_throttle_error = no_throttle_error48        self.filters_factory: FiltersFactory = filters_factory49        self.updates_handler = Handler(self, middleware_key='update')50        self.message_handlers = Handler(self, middleware_key='message')51        self.edited_message_handlers = Handler(self, middleware_key='edited_message')52        self.channel_post_handlers = Handler(self, middleware_key='channel_post')53        self.edited_channel_post_handlers = Handler(self, middleware_key='edited_channel_post')54        self.inline_query_handlers = Handler(self, middleware_key='inline_query')55        self.chosen_inline_result_handlers = Handler(self, middleware_key='chosen_inline_result')56        self.callback_query_handlers = Handler(self, middleware_key='callback_query')57        self.shipping_query_handlers = Handler(self, middleware_key='shipping_query')58        self.pre_checkout_query_handlers = Handler(self, middleware_key='pre_checkout_query')59        self.poll_handlers = Handler(self, middleware_key='poll')60        self.poll_answer_handlers = Handler(self, middleware_key='poll_answer')61        self.errors_handlers = Handler(self, once=False, middleware_key='error')62        self.middleware = MiddlewareManager(self)63        self.updates_handler.register(self.process_update)64        self._polling = False65        self._closed = True66        self._close_waiter = loop.create_future()67        self._setup_filters()68    def _setup_filters(self):69        filters_factory = self.filters_factory70        filters_factory.bind(StateFilter, exclude_event_handlers=[71            self.errors_handlers,72            self.poll_handlers,73            self.poll_answer_handlers,74        ])75        filters_factory.bind(ContentTypeFilter, event_handlers=[76            self.message_handlers,77            self.edited_message_handlers,78            self.channel_post_handlers,79            self.edited_channel_post_handlers,80        ]),81        filters_factory.bind(Command, event_handlers=[82            self.message_handlers,83            self.edited_message_handlers84        ])85        filters_factory.bind(Text, event_handlers=[86            self.message_handlers,87            self.edited_message_handlers,88            self.channel_post_handlers,89            self.edited_channel_post_handlers,90            self.callback_query_handlers,91            self.poll_handlers,92            self.inline_query_handlers,93        ])94        filters_factory.bind(HashTag, event_handlers=[95            self.message_handlers,96            self.edited_message_handlers,97            self.channel_post_handlers,98            self.edited_channel_post_handlers,99        ])100        filters_factory.bind(Regexp, event_handlers=[101            self.message_handlers,102            self.edited_message_handlers,103            self.channel_post_handlers,104            self.edited_channel_post_handlers,105            self.callback_query_handlers,106            self.poll_handlers,107            self.inline_query_handlers,108        ])109        filters_factory.bind(RegexpCommandsFilter, event_handlers=[110            self.message_handlers,111            self.edited_message_handlers,112        ])113        filters_factory.bind(ExceptionsFilter, event_handlers=[114            self.errors_handlers,115        ])116        filters_factory.bind(AdminFilter, event_handlers=[117            self.message_handlers, 118            self.edited_message_handlers,119            self.channel_post_handlers, 120            self.edited_channel_post_handlers,121            self.callback_query_handlers, 122            self.inline_query_handlers,123        ])124        filters_factory.bind(IDFilter, event_handlers=[125            self.message_handlers,126            self.edited_message_handlers,127            self.channel_post_handlers,128            self.edited_channel_post_handlers,129            self.callback_query_handlers,130            self.inline_query_handlers,131        ])132        filters_factory.bind(IsReplyFilter, event_handlers=[133            self.message_handlers,134            self.edited_message_handlers,135            self.channel_post_handlers,136            self.edited_channel_post_handlers,137        ])138        filters_factory.bind(IsSenderContact, event_handlers=[139            self.message_handlers,140            self.edited_message_handlers,141            self.channel_post_handlers,142            self.edited_channel_post_handlers,143        ])144    def __del__(self):145        self.stop_polling()146    async def skip_updates(self):147        """148        You can skip old incoming updates from queue.149        This method is not recommended to use if you use payments or you bot has high-load.150        :return: None151        """152        await self.bot.get_updates(offset=-1, timeout=1)153    async def process_updates(self, updates, fast: typing.Optional[bool] = True):154        """155        Process list of updates156        :param updates:157        :param fast:158        :return:159        """160        if fast:161            tasks = []162            for update in updates:163                tasks.append(self.updates_handler.notify(update))164            return await asyncio.gather(*tasks)165        results = []166        for update in updates:167            results.append(await self.updates_handler.notify(update))168        return results169    async def process_update(self, update: types.Update):170        """171        Process single update object172        :param update:173        :return:174        """175        types.Update.set_current(update)176        try:177            if update.message:178                types.User.set_current(update.message.from_user)179                types.Chat.set_current(update.message.chat)180                return await self.message_handlers.notify(update.message)181            if update.edited_message:182                types.User.set_current(update.edited_message.from_user)183                types.Chat.set_current(update.edited_message.chat)184                return await self.edited_message_handlers.notify(update.edited_message)185            if update.channel_post:186                types.Chat.set_current(update.channel_post.chat)187                return await self.channel_post_handlers.notify(update.channel_post)188            if update.edited_channel_post:189                types.Chat.set_current(update.edited_channel_post.chat)190                return await self.edited_channel_post_handlers.notify(update.edited_channel_post)191            if update.inline_query:192                types.User.set_current(update.inline_query.from_user)193                return await self.inline_query_handlers.notify(update.inline_query)194            if update.chosen_inline_result:195                types.User.set_current(update.chosen_inline_result.from_user)196                return await self.chosen_inline_result_handlers.notify(update.chosen_inline_result)197            if update.callback_query:198                if update.callback_query.message:199                    types.Chat.set_current(update.callback_query.message.chat)200                types.User.set_current(update.callback_query.from_user)201                return await self.callback_query_handlers.notify(update.callback_query)202            if update.shipping_query:203                types.User.set_current(update.shipping_query.from_user)204                return await self.shipping_query_handlers.notify(update.shipping_query)205            if update.pre_checkout_query:206                types.User.set_current(update.pre_checkout_query.from_user)207                return await self.pre_checkout_query_handlers.notify(update.pre_checkout_query)208            if update.poll:209                return await self.poll_handlers.notify(update.poll)210            if update.poll_answer:211                return await self.poll_answer_handlers.notify(update.poll_answer)212        except Exception as e:213            err = await self.errors_handlers.notify(update, e)214            if err:215                return err216            raise217    async def reset_webhook(self, check=True) -> bool:218        """219        Reset webhook220        :param check: check before deleting221        :return:222        """223        if check:224            wh = await self.bot.get_webhook_info()225            if not wh.url:226                return False227        return await self.bot.delete_webhook()228    async def start_polling(self,229                            timeout=20,230                            relax=0.1,231                            limit=None,232                            reset_webhook=None,233                            fast: typing.Optional[bool] = True,234                            error_sleep: int = 5):235        """236        Start long-polling237        :param timeout:238        :param relax:239        :param limit:240        :param reset_webhook:241        :param fast:242        :return:243        """244        if self._polling:245            raise RuntimeError('Polling already started')246        log.info('Start polling.')247        # context.set_value(MODE, LONG_POLLING)248        Dispatcher.set_current(self)249        Bot.set_current(self.bot)250        if reset_webhook is None:251            await self.reset_webhook(check=False)252        if reset_webhook:253            await self.reset_webhook(check=True)254        self._polling = True255        offset = None256        try:257            current_request_timeout = self.bot.timeout258            if current_request_timeout is not sentinel and timeout is not None:259                request_timeout = aiohttp.ClientTimeout(total=current_request_timeout.total + timeout or 1)260            else:261                request_timeout = None262            while self._polling:263                try:264                    with self.bot.request_timeout(request_timeout):265                        updates = await self.bot.get_updates(limit=limit, offset=offset, timeout=timeout)266                except asyncio.CancelledError:267                    break268                except:269                    log.exception('Cause exception while getting updates.')270                    await asyncio.sleep(error_sleep)271                    continue272                if updates:273                    log.debug(f"Received {len(updates)} updates.")274                    offset = updates[-1].update_id + 1275                    self.loop.create_task(self._process_polling_updates(updates, fast))276                if relax:277                    await asyncio.sleep(relax)278        finally:279            self._close_waiter.set_result(None)280            log.warning('Polling is stopped.')281    async def _process_polling_updates(self, updates, fast: typing.Optional[bool] = True):282        """283        Process updates received from long-polling.284        :param updates: list of updates.285        :param fast:286        """287        need_to_call = []288        for responses in itertools.chain.from_iterable(await self.process_updates(updates, fast)):289            for response in responses:290                if not isinstance(response, BaseResponse):291                    continue292                need_to_call.append(response.execute_response(self.bot))293        if need_to_call:294            try:295                asyncio.gather(*need_to_call)296            except TelegramAPIError:297                log.exception('Cause exception while processing updates.')298    def stop_polling(self):299        """300        Break long-polling process.301        :return:302        """303        if hasattr(self, '_polling') and self._polling:304            log.info('Stop polling...')305            self._polling = False306    async def wait_closed(self):307        """308        Wait for the long-polling to close309        :return:310        """311        await asyncio.shield(self._close_waiter, loop=self.loop)312    def is_polling(self):313        """314        Check if polling is enabled315        :return:316        """317        return self._polling318    def register_message_handler(self, callback, *custom_filters, commands=None, regexp=None, content_types=None,319                                 state=None, run_task=None, **kwargs):320        """321        Register handler for message322        .. code-block:: python3323            # This handler works only if state is None (by default).324            dp.register_message_handler(cmd_start, commands=['start', 'about'])325            dp.register_message_handler(entry_point, commands=['setup'])326            # This handler works only if current state is "first_step"327            dp.register_message_handler(step_handler_1, state="first_step")328            # If you want to handle all states by one handler, use `state="*"`.329            dp.register_message_handler(cancel_handler, commands=['cancel'], state="*")330            dp.register_message_handler(cancel_handler, lambda msg: msg.text.lower() == 'cancel', state="*")331        :param callback:332        :param commands: list of commands333        :param regexp: REGEXP334        :param content_types: List of content types.335        :param custom_filters: list of custom filters336        :param kwargs:337        :param state:338        :return: decorated function339        """340        filters_set = self.filters_factory.resolve(self.message_handlers,341                                                   *custom_filters,342                                                   commands=commands,343                                                   regexp=regexp,344                                                   content_types=content_types,345                                                   state=state,346                                                   **kwargs)347        self.message_handlers.register(self._wrap_async_task(callback, run_task), filters_set)348    def message_handler(self, *custom_filters, commands=None, regexp=None, content_types=None, state=None,349                        run_task=None, **kwargs):350        """351        Decorator for message handler352        Examples:353        Simple commands handler:354        .. code-block:: python3355            @dp.message_handler(commands=['start', 'welcome', 'about'])356            async def cmd_handler(message: types.Message):357        Filter messages by regular expression:358        .. code-block:: python3359            @dp.message_handler(rexexp='^[a-z]+-[0-9]+')360            async def msg_handler(message: types.Message):361        Filter messages by command regular expression:362        .. code-block:: python3363            @dp.message_handler(filters.RegexpCommandsFilter(regexp_commands=['item_([0-9]*)']))364            async def send_welcome(message: types.Message):365        Filter by content type:366        .. code-block:: python3367            @dp.message_handler(content_types=ContentType.PHOTO | ContentType.DOCUMENT)368            async def audio_handler(message: types.Message):369        Filter by custom function:370        .. code-block:: python3371            @dp.message_handler(lambda message: message.text and 'hello' in message.text.lower())372            async def text_handler(message: types.Message):373        Use multiple filters:374        .. code-block:: python3375            @dp.message_handler(commands=['command'], content_types=ContentType.TEXT)376            async def text_handler(message: types.Message):377        Register multiple filters set for one handler:378        .. code-block:: python3379            @dp.message_handler(commands=['command'])380            @dp.message_handler(lambda message: demojize(message.text) == ':new_moon_with_face:')381            async def text_handler(message: types.Message):382        This handler will be called if the message starts with '/command' OR is some emoji383        By default content_type is :class:`ContentType.TEXT`384        :param commands: list of commands385        :param regexp: REGEXP386        :param content_types: List of content types.387        :param custom_filters: list of custom filters388        :param kwargs:389        :param state:390        :param run_task: run callback in task (no wait results)391        :return: decorated function392        """393        def decorator(callback):394            self.register_message_handler(callback, *custom_filters,395                                          commands=commands, regexp=regexp, content_types=content_types,396                                          state=state, run_task=run_task, **kwargs)397            return callback398        return decorator399    def register_edited_message_handler(self, callback, *custom_filters, commands=None, regexp=None, content_types=None,400                                        state=None, run_task=None, **kwargs):401        """402        Register handler for edited message403        :param callback:404        :param commands: list of commands405        :param regexp: REGEXP406        :param content_types: List of content types.407        :param state:408        :param custom_filters: list of custom filters409        :param run_task: run callback in task (no wait results)410        :param kwargs:411        :return: decorated function412        """413        filters_set = self.filters_factory.resolve(self.edited_message_handlers,414                                                   *custom_filters,415                                                   commands=commands,416                                                   regexp=regexp,417                                                   content_types=content_types,418                                                   state=state,419                                                   **kwargs)420        self.edited_message_handlers.register(self._wrap_async_task(callback, run_task), filters_set)421    def edited_message_handler(self, *custom_filters, commands=None, regexp=None, content_types=None,422                               state=None, run_task=None, **kwargs):423        """424        Decorator for edited message handler425        You can use combination of different handlers426        .. code-block:: python3427            @dp.message_handler()428            @dp.edited_message_handler()429            async def msg_handler(message: types.Message):430        :param commands: list of commands431        :param regexp: REGEXP432        :param content_types: List of content types.433        :param state:434        :param custom_filters: list of custom filters435        :param run_task: run callback in task (no wait results)436        :param kwargs:437        :return: decorated function438        """439        def decorator(callback):440            self.register_edited_message_handler(callback, *custom_filters, commands=commands, regexp=regexp,441                                                 content_types=content_types, state=state, run_task=run_task, **kwargs)442            return callback443        return decorator444    def register_channel_post_handler(self, callback, *custom_filters, commands=None, regexp=None, content_types=None,445                                      state=None, run_task=None, **kwargs):446        """447        Register handler for channel post448        :param callback:449        :param commands: list of commands450        :param regexp: REGEXP451        :param content_types: List of content types.452        :param state:453        :param custom_filters: list of custom filters454        :param run_task: run callback in task (no wait results)455        :param kwargs:456        :return: decorated function457        """458        filters_set = self.filters_factory.resolve(self.channel_post_handlers,459                                                   *custom_filters,460                                                   commands=commands,461                                                   regexp=regexp,462                                                   content_types=content_types,463                                                   state=state,464                                                   **kwargs)465        self.channel_post_handlers.register(self._wrap_async_task(callback, run_task), filters_set)466    def channel_post_handler(self, *custom_filters, commands=None, regexp=None, content_types=None,467                             state=None, run_task=None, **kwargs):468        """469        Decorator for channel post handler470        :param commands: list of commands471        :param regexp: REGEXP472        :param content_types: List of content types.473        :param state:474        :param custom_filters: list of custom filters475        :param run_task: run callback in task (no wait results)476        :param kwargs:477        :return: decorated function478        """479        def decorator(callback):480            self.register_channel_post_handler(callback, *custom_filters, commands=commands, regexp=regexp,481                                               content_types=content_types, state=state, run_task=run_task, **kwargs)482            return callback483        return decorator484    def register_edited_channel_post_handler(self, callback, *custom_filters, commands=None, regexp=None,485                                             content_types=None, state=None, run_task=None, **kwargs):486        """487        Register handler for edited channel post488        :param callback:489        :param commands: list of commands490        :param regexp: REGEXP491        :param content_types: List of content types.492        :param state:493        :param custom_filters: list of custom filters494        :param run_task: run callback in task (no wait results)495        :param kwargs:496        :return: decorated function497        """498        filters_set = self.filters_factory.resolve(self.edited_message_handlers,499                                                   *custom_filters,500                                                   commands=commands,501                                                   regexp=regexp,502                                                   content_types=content_types,503                                                   state=state,504                                                   **kwargs)505        self.edited_channel_post_handlers.register(self._wrap_async_task(callback, run_task), filters_set)506    def edited_channel_post_handler(self, *custom_filters, commands=None, regexp=None, content_types=None,507                                    state=None, run_task=None, **kwargs):508        """509        Decorator for edited channel post handler510        :param commands: list of commands511        :param regexp: REGEXP512        :param content_types: List of content types.513        :param custom_filters: list of custom filters514        :param state:515        :param run_task: run callback in task (no wait results)516        :param kwargs:517        :return: decorated function518        """519        def decorator(callback):520            self.register_edited_channel_post_handler(callback, *custom_filters, commands=commands, regexp=regexp,521                                                      content_types=content_types, state=state, run_task=run_task,522                                                      **kwargs)523            return callback524        return decorator525    def register_inline_handler(self, callback, *custom_filters, state=None, run_task=None, **kwargs):526        """527        Register handler for inline query528        Example:529        .. code-block:: python3530            dp.register_inline_handler(some_inline_handler, lambda inline_query: True)531        :param callback:532        :param custom_filters: list of custom filters533        :param state:534        :param run_task: run callback in task (no wait results)535        :param kwargs:536        :return: decorated function537        """538        if custom_filters is None:539            custom_filters = []540        filters_set = self.filters_factory.resolve(self.inline_query_handlers,541                                                   *custom_filters,542                                                   state=state,543                                                   **kwargs)544        self.inline_query_handlers.register(self._wrap_async_task(callback, run_task), filters_set)545    def inline_handler(self, *custom_filters, state=None, run_task=None, **kwargs):546        """547        Decorator for inline query handler548        Example:549        .. code-block:: python3550            @dp.inline_handler(lambda inline_query: True)551            async def some_inline_handler(inline_query: types.InlineQuery)552        :param state:553        :param custom_filters: list of custom filters554        :param run_task: run callback in task (no wait results)555        :param kwargs:556        :return: decorated function557        """558        def decorator(callback):559            self.register_inline_handler(callback, *custom_filters, state=state, run_task=run_task, **kwargs)560            return callback561        return decorator562    def register_chosen_inline_handler(self, callback, *custom_filters, state=None, run_task=None, **kwargs):563        """564        Register handler for chosen inline query565        Example:566        .. code-block:: python3567            dp.register_chosen_inline_handler(some_chosen_inline_handler, lambda chosen_inline_query: True)568        :param callback:569        :param state:570        :param custom_filters:571        :param run_task: run callback in task (no wait results)572        :param kwargs:573        :return:574        """575        if custom_filters is None:576            custom_filters = []577        filters_set = self.filters_factory.resolve(self.chosen_inline_result_handlers,578                                                   *custom_filters,579                                                   state=state,580                                                   **kwargs)581        self.chosen_inline_result_handlers.register(self._wrap_async_task(callback, run_task), filters_set)582    def chosen_inline_handler(self, *custom_filters, state=None, run_task=None, **kwargs):583        """584        Decorator for chosen inline query handler585        Example:586        .. code-block:: python3587            @dp.chosen_inline_handler(lambda chosen_inline_query: True)588            async def some_chosen_inline_handler(chosen_inline_query: types.ChosenInlineResult)589        :param state:590        :param custom_filters:591        :param run_task: run callback in task (no wait results)592        :param kwargs:593        :return:594        """595        def decorator(callback):596            self.register_chosen_inline_handler(callback, *custom_filters, state=state, run_task=run_task, **kwargs)597            return callback598        return decorator599    def register_callback_query_handler(self, callback, *custom_filters, state=None, run_task=None, **kwargs):600        """601        Register handler for callback query602        Example:603        .. code-block:: python3604            dp.register_callback_query_handler(some_callback_handler, lambda callback_query: True)605        :param callback:606        :param state:607        :param custom_filters:608        :param run_task: run callback in task (no wait results)609        :param kwargs:610        """611        filters_set = self.filters_factory.resolve(self.callback_query_handlers,612                                                   *custom_filters,613                                                   state=state,614                                                   **kwargs)615        self.callback_query_handlers.register(self._wrap_async_task(callback, run_task), filters_set)616    def callback_query_handler(self, *custom_filters, state=None, run_task=None, **kwargs):617        """618        Decorator for callback query handler619        Example:620        .. code-block:: python3621            @dp.callback_query_handler(lambda callback_query: True)622            async def some_callback_handler(callback_query: types.CallbackQuery)623        :param state:624        :param custom_filters:625        :param run_task: run callback in task (no wait results)626        :param kwargs:627        """628        def decorator(callback):629            self.register_callback_query_handler(callback, *custom_filters, state=state, run_task=run_task, **kwargs)630            return callback631        return decorator632    def register_shipping_query_handler(self, callback, *custom_filters, state=None, run_task=None,633                                        **kwargs):634        """635        Register handler for shipping query636        Example:637        .. code-block:: python3638            dp.register_shipping_query_handler(some_shipping_query_handler, lambda shipping_query: True)639        :param callback:640        :param state:641        :param custom_filters:642        :param run_task: run callback in task (no wait results)643        :param kwargs:644        """645        filters_set = self.filters_factory.resolve(self.shipping_query_handlers,646                                                   *custom_filters,647                                                   state=state,648                                                   **kwargs)649        self.shipping_query_handlers.register(self._wrap_async_task(callback, run_task), filters_set)650    def shipping_query_handler(self, *custom_filters, state=None, run_task=None, **kwargs):651        """652        Decorator for shipping query handler653        Example:654        .. code-block:: python3655            @dp.shipping_query_handler(lambda shipping_query: True)656            async def some_shipping_query_handler(shipping_query: types.ShippingQuery)657        :param state:658        :param custom_filters:659        :param run_task: run callback in task (no wait results)660        :param kwargs:661        """662        def decorator(callback):663            self.register_shipping_query_handler(callback, *custom_filters, state=state, run_task=run_task, **kwargs)664            return callback665        return decorator666    def register_pre_checkout_query_handler(self, callback, *custom_filters, state=None, run_task=None, **kwargs):667        """668        Register handler for pre-checkout query669        Example:670        .. code-block:: python3671            dp.register_pre_checkout_query_handler(some_pre_checkout_query_handler, lambda shipping_query: True)672        :param callback:673        :param state:674        :param custom_filters:675        :param run_task: run callback in task (no wait results)676        :param kwargs:677        """678        filters_set = self.filters_factory.resolve(self.pre_checkout_query_handlers,679                                                   *custom_filters,680                                                   state=state,681                                                   **kwargs)682        self.pre_checkout_query_handlers.register(self._wrap_async_task(callback, run_task), filters_set)683    def pre_checkout_query_handler(self, *custom_filters, state=None, run_task=None, **kwargs):684        """685        Decorator for pre-checkout query handler686        Example:687        .. code-block:: python3688            @dp.pre_checkout_query_handler(lambda shipping_query: True)689            async def some_pre_checkout_query_handler(shipping_query: types.ShippingQuery)690        :param state:691        :param custom_filters:692        :param run_task: run callback in task (no wait results)693        :param kwargs:694        """695        def decorator(callback):696            self.register_pre_checkout_query_handler(callback, *custom_filters, state=state, run_task=run_task,697                                                     **kwargs)698            return callback699        return decorator700    def register_poll_handler(self, callback, *custom_filters, run_task=None, **kwargs):701        """702        Register handler for poll703        704        Example:705        .. code-block:: python3706            dp.register_poll_handler(some_poll_handler)707        :param callback:708        :param custom_filters:709        :param run_task: run callback in task (no wait results)710        :param kwargs:711        """712        filters_set = self.filters_factory.resolve(self.poll_handlers,713                                                   *custom_filters,714                                                   **kwargs)715        self.poll_handlers.register(self._wrap_async_task(callback, run_task), filters_set)716    def poll_handler(self, *custom_filters, run_task=None, **kwargs):717        """718        Decorator for poll handler719        Example:720        .. code-block:: python3721            @dp.poll_handler()722            async def some_poll_handler(poll: types.Poll)723        :param custom_filters:724        :param run_task: run callback in task (no wait results)725        :param kwargs:726        """727        728        def decorator(callback):729            self.register_poll_handler(callback, *custom_filters, run_task=run_task,730                                       **kwargs)731            return callback732        return decorator733    734    def register_poll_answer_handler(self, callback, *custom_filters, run_task=None, **kwargs):735        """736        Register handler for poll_answer737        738        Example:739        .. code-block:: python3740            dp.register_poll_answer_handler(some_poll_answer_handler)741        :param callback:742        :param custom_filters:743        :param run_task: run callback in task (no wait results)744        :param kwargs:745        """746        filters_set = self.filters_factory.resolve(self.poll_answer_handlers,747                                                   *custom_filters,748                                                   **kwargs)749        self.poll_answer_handlers.register(self._wrap_async_task(callback, run_task), filters_set)750    751    def poll_answer_handler(self, *custom_filters, run_task=None, **kwargs):752        """753        Decorator for poll_answer handler754        Example:755        .. code-block:: python3756            @dp.poll_answer_handler()757            async def some_poll_answer_handler(poll_answer: types.PollAnswer)758        :param custom_filters:759        :param run_task: run callback in task (no wait results)760        :param kwargs:761        """762        def decorator(callback):763            self.register_poll_answer_handler(callback, *custom_filters, run_task=run_task,764                                       **kwargs)765            return callback766        return decorator767    def register_errors_handler(self, callback, *custom_filters, exception=None, run_task=None, **kwargs):768        """769        Register handler for errors770        :param callback:771        :param exception: you can make handler for specific errors type772        :param run_task: run callback in task (no wait results)773        """774        filters_set = self.filters_factory.resolve(self.errors_handlers,775                                                   *custom_filters,776                                                   exception=exception,777                                                   **kwargs)778        self.errors_handlers.register(self._wrap_async_task(callback, run_task), filters_set)779    def errors_handler(self, *custom_filters, exception=None, run_task=None, **kwargs):780        """781        Decorator for errors handler782        :param exception: you can make handler for specific errors type783        :param run_task: run callback in task (no wait results)784        :return:785        """786        def decorator(callback):787            self.register_errors_handler(self._wrap_async_task(callback, run_task),788                                         *custom_filters, exception=exception, **kwargs)789            return callback790        return decorator791    def current_state(self, *,792                      chat: typing.Union[str, int, None] = None,793                      user: typing.Union[str, int, None] = None) -> FSMContext:794        """795        Get current state for user in chat as context796        .. code-block:: python3797            with dp.current_state(chat=message.chat.id, user=message.user.id) as state:798                pass799            state = dp.current_state()800            state.set_state('my_state')801        :param chat:802        :param user:803        :return:804        """805        if chat is None:806            chat_obj = types.Chat.get_current()807            chat = chat_obj.id if chat_obj else None808        if user is None:809            user_obj = types.User.get_current()810            user = user_obj.id if user_obj else None811        return FSMContext(storage=self.storage, chat=chat, user=user)812    @renamed_argument(old_name='user', new_name='user_id', until_version='3.0', stacklevel=3)813    @renamed_argument(old_name='chat', new_name='chat_id', until_version='3.0', stacklevel=4)814    async def throttle(self, key, *, rate=None, user_id=None, chat_id=None, no_error=None) -> bool:815        """816        Execute throttling manager.817        Returns True if limit has not exceeded otherwise raises ThrottleError or returns False818        :param key: key in storage819        :param rate: limit (by default is equal to default rate limit)820        :param user_id: user id821        :param chat_id: chat id822        :param no_error: return boolean value instead of raising error823        :return: bool824        """825        if not self.storage.has_bucket():826            raise RuntimeError('This storage does not provide Leaky Bucket')827        if no_error is None:828            no_error = self.no_throttle_error829        if rate is None:830            rate = self.throttling_rate_limit831        if user_id is None and chat_id is None:832            user_id = types.User.get_current().id833            chat_id = types.Chat.get_current().id834        # Detect current time835        now = time.time()836        bucket = await self.storage.get_bucket(chat=chat_id, user=user_id)837        # Fix bucket838        if bucket is None:839            bucket = {key: {}}840        if key not in bucket:841            bucket[key] = {}842        data = bucket[key]843        # Calculate844        called = data.get(LAST_CALL, now)845        delta = now - called846        result = delta >= rate or delta <= 0847        # Save results848        data[RESULT] = result849        data[RATE_LIMIT] = rate850        data[LAST_CALL] = now851        data[DELTA] = delta852        if not result:853            data[EXCEEDED_COUNT] += 1854        else:855            data[EXCEEDED_COUNT] = 1856        bucket[key].update(data)857        await self.storage.set_bucket(chat=chat_id, user=user_id, bucket=bucket)858        if not result and not no_error:859            # Raise if it is allowed860            raise Throttled(key=key, chat=chat_id, user=user_id, **data)861        return result862    @renamed_argument(old_name='user', new_name='user_id', until_version='3.0', stacklevel=3)863    @renamed_argument(old_name='chat', new_name='chat_id', until_version='3.0', stacklevel=4)864    async def check_key(self, key, chat_id=None, user_id=None):865        """866        Get information about key in bucket867        :param key:868        :param chat_id:869        :param user_id:870        :return:871        """872        if not self.storage.has_bucket():873            raise RuntimeError('This storage does not provide Leaky Bucket')874        if user_id is None and chat_id is None:875            user_id = types.User.get_current()876            chat_id = types.Chat.get_current()877        bucket = await self.storage.get_bucket(chat=chat_id, user=user_id)878        data = bucket.get(key, {})879        return Throttled(key=key, chat=chat_id, user=user_id, **data)880    @renamed_argument(old_name='user', new_name='user_id', until_version='3.0', stacklevel=3)881    @renamed_argument(old_name='chat', new_name='chat_id', until_version='3.0', stacklevel=4)882    async def release_key(self, key, chat_id=None, user_id=None):883        """884        Release blocked key885        :param key:886        :param chat_id:887        :param user_id:888        :return:889        """890        if not self.storage.has_bucket():891            raise RuntimeError('This storage does not provide Leaky Bucket')892        if user_id is None and chat_id is None:893            user_id = types.User.get_current()894            chat_id = types.Chat.get_current()895        bucket = await self.storage.get_bucket(chat=chat_id, user=user_id)896        if bucket and key in bucket:897            del bucket['key']898            await self.storage.set_bucket(chat=chat_id, user=user_id, bucket=bucket)899            return True900        return False901    def async_task(self, func):902        """903        Execute handler as task and return None.904        Use this decorator for slow handlers (with timeouts)905        .. code-block:: python3906            @dp.message_handler(commands=['command'])907            @dp.async_task908            async def cmd_with_timeout(message: types.Message):909                await asyncio.sleep(120)910                return SendMessage(message.chat.id, 'KABOOM').reply(message)911        :param func:912        :return:913        """914        def process_response(task):915            try:916                response = task.result()917            except Exception as e:918                self.loop.create_task(919                    self.errors_handlers.notify(types.Update.get_current(), e))920            else:921                if isinstance(response, BaseResponse):922                    self.loop.create_task(response.execute_response(self.bot))923        @functools.wraps(func)924        async def wrapper(*args, **kwargs):925            task = self.loop.create_task(func(*args, **kwargs))926            task.add_done_callback(process_response)927        return wrapper928    def _wrap_async_task(self, callback, run_task=None) -> callable:929        if run_task is None:930            run_task = self.run_tasks_by_default931        if run_task:932            return self.async_task(callback)933        return callback934    def throttled(self, on_throttled: typing.Optional[typing.Callable] = None,935                  key=None, rate=None,936                  user_id=None, chat_id=None):937        """938        Meta-decorator for throttling.939        Invokes on_throttled if the handler was throttled.940        Example:941        .. code-block:: python3942            async def handler_throttled(message: types.Message, **kwargs):943                await message.answer("Throttled!")944            @dp.throttled(handler_throttled)945            async def some_handler(message: types.Message):946                await message.answer("Didn't throttled!")947        :param on_throttled: the callable object that should be either a function or return a coroutine948        :param key: key in storage949        :param rate: limit (by default is equal to default rate limit)950        :param user_id: user id951        :param chat_id: chat id952        :return: decorator953        """954        def decorator(func):955            @functools.wraps(func)956            async def wrapped(*args, **kwargs):957                is_not_throttled = await self.throttle(key if key is not None else func.__name__,958                                                       rate=rate,959                                                       user_id=user_id, chat_id=chat_id,960                                                       no_error=True)961                if is_not_throttled:962                    return await func(*args, **kwargs)963                else:964                    kwargs.update(965                        {966                            'rate': rate,967                            'key': key,968                            'user_id': user_id,969                            'chat_id': chat_id970                        }971                    )  # update kwargs with parameters which were given to throttled972                    if on_throttled:973                        if asyncio.iscoroutinefunction(on_throttled):974                            await on_throttled(*args, **kwargs)975                        else:976                            kwargs.update(977                                {978                                    'loop': asyncio.get_running_loop()979                                }980                            )981                            partial_func = functools.partial(on_throttled, *args, **kwargs)982                            asyncio.get_running_loop().run_in_executor(None,983                                                                       partial_func984                                                                       )985            return wrapped...

__init__.py

Source:__init__.py

1import asyncio2import functools3import itertools4import logging5import time6import typing7from .filters import CommandsFilter, ContentTypeFilter, ExceptionsFilter, RegexpFilter, \8    USER_STATE, generate_default_filters9from .handler import CancelHandler, Handler, SkipHandler10from .middlewares import MiddlewareManager11from .storage import BaseStorage, DELTA, DisabledStorage, EXCEEDED_COUNT, FSMContext, \12    LAST_CALL, RATE_LIMIT, RESULT13from .webhook import BaseResponse14from ..bot import Bot15from ..types.message import ContentType16from ..utils import context17from ..utils.deprecated import deprecated18from ..utils.exceptions import NetworkError, TelegramAPIError, Throttled19log = logging.getLogger(__name__)20MODE = 'MODE'21LONG_POLLING = 'long-polling'22UPDATE_OBJECT = 'update_object'23DEFAULT_RATE_LIMIT = .124class Dispatcher:25    """26    Simple Updates dispatcher27    It will process incoming updates: messages, edited messages, channel posts, edited channel posts,28    inline queries, chosen inline results, callback queries, shipping queries, pre-checkout queries.29    """30    def __init__(self, bot, loop=None, storage: typing.Optional[BaseStorage] = None,31                 run_tasks_by_default: bool = False,32                 throttling_rate_limit=DEFAULT_RATE_LIMIT, no_throttle_error=False):33        if loop is None:34            loop = bot.loop35        if storage is None:36            storage = DisabledStorage()37        self.bot: Bot = bot38        self.loop = loop39        self.storage = storage40        self.run_tasks_by_default = run_tasks_by_default41        self.throttling_rate_limit = throttling_rate_limit42        self.no_throttle_error = no_throttle_error43        self.last_update_id = 044        self.updates_handler = Handler(self, middleware_key='update')45        self.message_handlers = Handler(self, middleware_key='message')46        self.edited_message_handlers = Handler(self, middleware_key='edited_message')47        self.channel_post_handlers = Handler(self, middleware_key='channel_post')48        self.edited_channel_post_handlers = Handler(self, middleware_key='edited_channel_post')49        self.inline_query_handlers = Handler(self, middleware_key='inline_query')50        self.chosen_inline_result_handlers = Handler(self, middleware_key='chosen_inline_result')51        self.callback_query_handlers = Handler(self, middleware_key='callback_query')52        self.shipping_query_handlers = Handler(self, middleware_key='shipping_query')53        self.pre_checkout_query_handlers = Handler(self, middleware_key='pre_checkout_query')54        self.errors_handlers = Handler(self, once=False, middleware_key='error')55        self.middleware = MiddlewareManager(self)56        self.updates_handler.register(self.process_update)57        self._polling = False58        self._closed = True59        self._close_waiter = loop.create_future()60    def __del__(self):61        self.stop_polling()62    @property63    def data(self):64        return self.bot.data65    def __setitem__(self, key, value):66        self.bot.data[key] = value67    def __getitem__(self, item):68        return self.bot.data[item]69    def get(self, key, default=None):70        return self.bot.data.get(key, default)71    async def skip_updates(self):72        """73        You can skip old incoming updates from queue.74        This method is not recommended to use if you use payments or you bot has high-load.75        :return: count of skipped updates76        """77        total = 078        updates = await self.bot.get_updates(offset=self.last_update_id, timeout=1)79        while updates:80            total += len(updates)81            for update in updates:82                if update.update_id > self.last_update_id:83                    self.last_update_id = update.update_id84            updates = await self.bot.get_updates(offset=self.last_update_id + 1, timeout=1)85        return total86    async def process_updates(self, updates):87        """88        Process list of updates89        :param updates:90        :return:91        """92        tasks = []93        for update in updates:94            tasks.append(self.updates_handler.notify(update))95        return await asyncio.gather(*tasks)96    async def process_update(self, update):97        """98        Process single update object99        :param update:100        :return:101        """102        self.last_update_id = update.update_id103        context.set_value(UPDATE_OBJECT, update)104        try:105            if update.message:106                state = await self.storage.get_state(chat=update.message.chat.id,107                                                     user=update.message.from_user.id)108                context.update_state(chat=update.message.chat.id,109                                     user=update.message.from_user.id,110                                     state=state)111                return await self.message_handlers.notify(update.message)112            if update.edited_message:113                state = await self.storage.get_state(chat=update.edited_message.chat.id,114                                                     user=update.edited_message.from_user.id)115                context.update_state(chat=update.edited_message.chat.id,116                                     user=update.edited_message.from_user.id,117                                     state=state)118                return await self.edited_message_handlers.notify(update.edited_message)119            if update.channel_post:120                state = await self.storage.get_state(chat=update.channel_post.chat.id)121                context.update_state(chat=update.channel_post.chat.id,122                                     state=state)123                return await self.channel_post_handlers.notify(update.channel_post)124            if update.edited_channel_post:125                state = await self.storage.get_state(chat=update.edited_channel_post.chat.id)126                context.update_state(chat=update.edited_channel_post.chat.id,127                                     state=state)128                return await self.edited_channel_post_handlers.notify(update.edited_channel_post)129            if update.inline_query:130                state = await self.storage.get_state(user=update.inline_query.from_user.id)131                context.update_state(user=update.inline_query.from_user.id,132                                     state=state)133                return await self.inline_query_handlers.notify(update.inline_query)134            if update.chosen_inline_result:135                state = await self.storage.get_state(user=update.chosen_inline_result.from_user.id)136                context.update_state(user=update.chosen_inline_result.from_user.id,137                                     state=state)138                return await self.chosen_inline_result_handlers.notify(update.chosen_inline_result)139            if update.callback_query:140                state = await self.storage.get_state(141                    chat=update.callback_query.message.chat.id if update.callback_query.message else None,142                    user=update.callback_query.from_user.id)143                context.update_state(user=update.callback_query.from_user.id,144                                     state=state)145                return await self.callback_query_handlers.notify(update.callback_query)146            if update.shipping_query:147                state = await self.storage.get_state(user=update.shipping_query.from_user.id)148                context.update_state(user=update.shipping_query.from_user.id,149                                     state=state)150                return await self.shipping_query_handlers.notify(update.shipping_query)151            if update.pre_checkout_query:152                state = await self.storage.get_state(user=update.pre_checkout_query.from_user.id)153                context.update_state(user=update.pre_checkout_query.from_user.id,154                                     state=state)155                return await self.pre_checkout_query_handlers.notify(update.pre_checkout_query)156        except Exception as e:157            err = await self.errors_handlers.notify(self, update, e)158            if err:159                return err160            raise161    async def reset_webhook(self, check=True) -> bool:162        """163        Reset webhook164        :param check: check before deleting165        :return:166        """167        if check:168            wh = await self.bot.get_webhook_info()169            if not wh.url:170                return False171        return await self.bot.delete_webhook()172    @deprecated('The old method was renamed to `start_polling`')173    async def start_pooling(self, *args, **kwargs):174        """175        Start long-lopping176        :param args:177        :param kwargs:178        :return:179        """180        return await self.start_polling(*args, **kwargs)181    async def start_polling(self, timeout=20, relax=0.1, limit=None, reset_webhook=None):182        """183        Start long-polling184        :param timeout:185        :param relax:186        :param limit:187        :param reset_webhook:188        :return:189        """190        if self._polling:191            raise RuntimeError('Polling already started')192        log.info('Start polling.')193        context.set_value(MODE, LONG_POLLING)194        context.set_value('dispatcher', self)195        context.set_value('bot', self.bot)196        if reset_webhook is None:197            await self.reset_webhook(check=False)198        if reset_webhook:199            await self.reset_webhook(check=True)200        self._polling = True201        offset = None202        try:203            while self._polling:204                try:205                    updates = await self.bot.get_updates(limit=limit, offset=offset, timeout=timeout)206                except NetworkError:207                    log.exception('Cause exception while getting updates.')208                    await asyncio.sleep(15)209                    continue210                if updates:211                    log.debug(f"Received {len(updates)} updates.")212                    offset = updates[-1].update_id + 1213                    self.loop.create_task(self._process_polling_updates(updates))214                if relax:215                    await asyncio.sleep(relax)216        finally:217            self._close_waiter.set_result(None)218            log.warning('Polling is stopped.')219    async def _process_polling_updates(self, updates):220        """221        Process updates received from long-polling.222        :param updates: list of updates.223        """224        need_to_call = []225        for responses in itertools.chain.from_iterable(await self.process_updates(updates)):226            for response in responses:227                if not isinstance(response, BaseResponse):228                    continue229                need_to_call.append(response.execute_response(self.bot))230        if need_to_call:231            try:232                asyncio.gather(*need_to_call)233            except TelegramAPIError:234                log.exception('Cause exception while processing updates.')235    @deprecated('The old method was renamed to `stop_polling`')236    def stop_pooling(self):237        return self.stop_polling()238    def stop_polling(self):239        """240        Break long-polling process.241        :return:242        """243        if self._polling:244            log.info('Stop polling...')245            self._polling = False246    async def wait_closed(self):247        """248        Wait for the long-polling to close249        :return:250        """251        await asyncio.shield(self._close_waiter, loop=self.loop)252    @deprecated('The old method was renamed to `is_polling`')253    def is_pooling(self):254        return self.is_polling()255    def is_polling(self):256        """257        Check if polling is enabled258        :return:259        """260        return self._polling261    def register_message_handler(self, callback, *, commands=None, regexp=None, content_types=None, func=None,262                                 state=None, custom_filters=None, run_task=None, **kwargs):263        """264        Register handler for message265        .. code-block:: python3266            # This handler works only if state is None (by default).267            dp.register_message_handler(cmd_start, commands=['start', 'about'])268            dp.register_message_handler(entry_point, commands=['setup'])269            # This handler works only if current state is "first_step"270            dp.register_message_handler(step_handler_1, state="first_step")271            # If you want to handle all states by one handler, use `state="*"`.272            dp.register_message_handler(cancel_handler, commands=['cancel'], state="*")273            dp.register_message_handler(cancel_handler, func=lambda msg: msg.text.lower() == 'cancel', state="*")274        :param callback:275        :param commands: list of commands276        :param regexp: REGEXP277        :param content_types: List of content types.278        :param func: custom any callable object279        :param custom_filters: list of custom filters280        :param kwargs:281        :param state:282        :return: decorated function283        """284        if content_types is None:285            content_types = ContentType.TEXT286        if custom_filters is None:287            custom_filters = []288        filters_set = generate_default_filters(self,289                                               *custom_filters,290                                               commands=commands,291                                               regexp=regexp,292                                               content_types=content_types,293                                               func=func,294                                               state=state,295                                               **kwargs)296        self.message_handlers.register(self._wrap_async_task(callback, run_task), filters_set)297    def message_handler(self, *custom_filters, commands=None, regexp=None, content_types=None, func=None, state=None,298                        run_task=None, **kwargs):299        """300        Decorator for message handler301        Examples:302        Simple commands handler:303        .. code-block:: python3304            @dp.message_handler(commands=['start', 'welcome', 'about'])305            async def cmd_handler(message: types.Message):306        Filter messages by regular expression:307        .. code-block:: python3308            @dp.message_handler(rexexp='^[a-z]+-[0-9]+')309            async def msg_handler(message: types.Message):310        Filter messages by command regular expression:311        .. code-block:: python3312            @dp.message_handler(filters.RegexpCommandsFilter(regexp_commands=['item_([0-9]*)']))313            async def send_welcome(message: types.Message):314        Filter by content type:315        .. code-block:: python3316            @dp.message_handler(content_types=ContentType.PHOTO | ContentType.DOCUMENT)317            async def audio_handler(message: types.Message):318        Filter by custom function:319        .. code-block:: python3320            @dp.message_handler(func=lambda message: message.text and 'hello' in message.text.lower())321            async def text_handler(message: types.Message):322        Use multiple filters:323        .. code-block:: python3324            @dp.message_handler(commands=['command'], content_types=ContentType.TEXT)325            async def text_handler(message: types.Message):326        Register multiple filters set for one handler:327        .. code-block:: python3328            @dp.message_handler(commands=['command'])329            @dp.message_handler(func=lambda message: demojize(message.text) == ':new_moon_with_face:')330            async def text_handler(message: types.Message):331        This handler will be called if the message starts with '/command' OR is some emoji332        By default content_type is :class:`ContentType.TEXT`333        :param commands: list of commands334        :param regexp: REGEXP335        :param content_types: List of content types.336        :param func: custom any callable object337        :param custom_filters: list of custom filters338        :param kwargs:339        :param state:340        :param run_task: run callback in task (no wait results)341        :return: decorated function342        """343        def decorator(callback):344            self.register_message_handler(callback,345                                          commands=commands, regexp=regexp, content_types=content_types,346                                          func=func, state=state, custom_filters=custom_filters, run_task=run_task,347                                          **kwargs)348            return callback349        return decorator350    def register_edited_message_handler(self, callback, *, commands=None, regexp=None, content_types=None, func=None,351                                        state=None, custom_filters=None, run_task=None, **kwargs):352        """353        Register handler for edited message354        :param callback:355        :param commands: list of commands356        :param regexp: REGEXP357        :param content_types: List of content types.358        :param func: custom any callable object359        :param state:360        :param custom_filters: list of custom filters361        :param run_task: run callback in task (no wait results)362        :param kwargs:363        :return: decorated function364        """365        if content_types is None:366            content_types = ContentType.TEXT367        if custom_filters is None:368            custom_filters = []369        filters_set = generate_default_filters(self,370                                               *custom_filters,371                                               commands=commands,372                                               regexp=regexp,373                                               content_types=content_types,374                                               func=func,375                                               state=state,376                                               **kwargs)377        self.edited_message_handlers.register(self._wrap_async_task(callback, run_task), filters_set)378    def edited_message_handler(self, *custom_filters, commands=None, regexp=None, content_types=None, func=None,379                               state=None, run_task=None, **kwargs):380        """381        Decorator for edited message handler382        You can use combination of different handlers383        .. code-block:: python3384            @dp.message_handler()385            @dp.edited_message_handler()386            async def msg_handler(message: types.Message):387        :param commands: list of commands388        :param regexp: REGEXP389        :param content_types: List of content types.390        :param func: custom any callable object391        :param state:392        :param custom_filters: list of custom filters393        :param run_task: run callback in task (no wait results)394        :param kwargs:395        :return: decorated function396        """397        def decorator(callback):398            self.register_edited_message_handler(callback, commands=commands, regexp=regexp,399                                                 content_types=content_types, func=func, state=state,400                                                 custom_filters=custom_filters, run_task=run_task, **kwargs)401            return callback402        return decorator403    def register_channel_post_handler(self, callback, *, commands=None, regexp=None, content_types=None, func=None,404                                      state=None, custom_filters=None, run_task=None, **kwargs):405        """406        Register handler for channel post407        :param callback:408        :param commands: list of commands409        :param regexp: REGEXP410        :param content_types: List of content types.411        :param func: custom any callable object412        :param state:413        :param custom_filters: list of custom filters414        :param run_task: run callback in task (no wait results)415        :param kwargs:416        :return: decorated function417        """418        if content_types is None:419            content_types = ContentType.TEXT420        if custom_filters is None:421            custom_filters = []422        filters_set = generate_default_filters(self,423                                               *custom_filters,424                                               commands=commands,425                                               regexp=regexp,426                                               content_types=content_types,427                                               func=func,428                                               state=state,429                                               **kwargs)430        self.channel_post_handlers.register(self._wrap_async_task(callback, run_task), filters_set)431    def channel_post_handler(self, *custom_filters, commands=None, regexp=None, content_types=None, func=None,432                             state=None, run_task=None, **kwargs):433        """434        Decorator for channel post handler435        :param commands: list of commands436        :param regexp: REGEXP437        :param content_types: List of content types.438        :param func: custom any callable object439        :param state:440        :param custom_filters: list of custom filters441        :param run_task: run callback in task (no wait results)442        :param kwargs:443        :return: decorated function444        """445        def decorator(callback):446            self.register_channel_post_handler(callback, commands=commands, regexp=regexp, content_types=content_types,447                                               func=func, state=state, custom_filters=custom_filters,448                                               run_task=run_task, **kwargs)449            return callback450        return decorator451    def register_edited_channel_post_handler(self, callback, *, commands=None, regexp=None, content_types=None,452                                             func=None, state=None, custom_filters=None, run_task=None, **kwargs):453        """454        Register handler for edited channel post455        :param callback:456        :param commands: list of commands457        :param regexp: REGEXP458        :param content_types: List of content types.459        :param func: custom any callable object460        :param state:461        :param custom_filters: list of custom filters462        :param run_task: run callback in task (no wait results)463        :param kwargs:464        :return: decorated function465        """466        if content_types is None:467            content_types = ContentType.TEXT468        if custom_filters is None:469            custom_filters = []470        filters_set = generate_default_filters(self,471                                               *custom_filters,472                                               commands=commands,473                                               regexp=regexp,474                                               content_types=content_types,475                                               func=func,476                                               state=state,477                                               **kwargs)478        self.edited_channel_post_handlers.register(self._wrap_async_task(callback, run_task), filters_set)479    def edited_channel_post_handler(self, *custom_filters, commands=None, regexp=None, content_types=None, func=None,480                                    state=None, run_task=None, **kwargs):481        """482        Decorator for edited channel post handler483        :param commands: list of commands484        :param regexp: REGEXP485        :param content_types: List of content types.486        :param func: custom any callable object487        :param custom_filters: list of custom filters488        :param state:489        :param run_task: run callback in task (no wait results)490        :param kwargs:491        :return: decorated function492        """493        def decorator(callback):494            self.register_edited_channel_post_handler(callback, commands=commands, regexp=regexp,495                                                      content_types=content_types, func=func, state=state,496                                                      custom_filters=custom_filters, run_task=run_task, **kwargs)497            return callback498        return decorator499    def register_inline_handler(self, callback, *, func=None, state=None, custom_filters=None, run_task=None, **kwargs):500        """501        Register handler for inline query502        Example:503        .. code-block:: python3504            dp.register_inline_handler(some_inline_handler, func=lambda inline_query: True)505        :param callback:506        :param func: custom any callable object507        :param custom_filters: list of custom filters508        :param state:509        :param run_task: run callback in task (no wait results)510        :param kwargs:511        :return: decorated function512        """513        if custom_filters is None:514            custom_filters = []515        filters_set = generate_default_filters(self,516                                               *custom_filters,517                                               func=func,518                                               state=state,519                                               **kwargs)520        self.inline_query_handlers.register(self._wrap_async_task(callback, run_task), filters_set)521    def inline_handler(self, *custom_filters, func=None, state=None, run_task=None, **kwargs):522        """523        Decorator for inline query handler524        Example:525        .. code-block:: python3526            @dp.inline_handler(func=lambda inline_query: True)527            async def some_inline_handler(inline_query: types.InlineQuery)528        :param func: custom any callable object529        :param state:530        :param custom_filters: list of custom filters531        :param run_task: run callback in task (no wait results)532        :param kwargs:533        :return: decorated function534        """535        def decorator(callback):536            self.register_inline_handler(callback, func=func, state=state, custom_filters=custom_filters,537                                         run_task=run_task, **kwargs)538            return callback539        return decorator540    def register_chosen_inline_handler(self, callback, *, func=None, state=None, custom_filters=None, run_task=None,541                                       **kwargs):542        """543        Register handler for chosen inline query544        Example:545        .. code-block:: python3546            dp.register_chosen_inline_handler(some_chosen_inline_handler, func=lambda chosen_inline_query: True)547        :param callback:548        :param func: custom any callable object549        :param state:550        :param custom_filters:551        :param run_task: run callback in task (no wait results)552        :param kwargs:553        :return:554        """555        if custom_filters is None:556            custom_filters = []557        filters_set = generate_default_filters(self,558                                               *custom_filters,559                                               func=func,560                                               state=state,561                                               **kwargs)562        self.chosen_inline_result_handlers.register(self._wrap_async_task(callback, run_task), filters_set)563    def chosen_inline_handler(self, *custom_filters, func=None, state=None, run_task=None, **kwargs):564        """565        Decorator for chosen inline query handler566        Example:567        .. code-block:: python3568            @dp.chosen_inline_handler(func=lambda chosen_inline_query: True)569            async def some_chosen_inline_handler(chosen_inline_query: types.ChosenInlineResult)570        :param func: custom any callable object571        :param state:572        :param custom_filters:573        :param run_task: run callback in task (no wait results)574        :param kwargs:575        :return:576        """577        def decorator(callback):578            self.register_chosen_inline_handler(callback, func=func, state=state, custom_filters=custom_filters,579                                                run_task=run_task, **kwargs)580            return callback581        return decorator582    def register_callback_query_handler(self, callback, *, func=None, state=None, custom_filters=None, run_task=None,583                                        **kwargs):584        """585        Register handler for callback query586        Example:587        .. code-block:: python3588            dp.register_callback_query_handler(some_callback_handler, func=lambda callback_query: True)589        :param callback:590        :param func: custom any callable object591        :param state:592        :param custom_filters:593        :param run_task: run callback in task (no wait results)594        :param kwargs:595        """596        if custom_filters is None:597            custom_filters = []598        filters_set = generate_default_filters(self,599                                               *custom_filters,600                                               func=func,601                                               state=state,602                                               **kwargs)603        self.callback_query_handlers.register(self._wrap_async_task(callback, run_task), filters_set)604    def callback_query_handler(self, *custom_filters, func=None, state=None, run_task=None, **kwargs):605        """606        Decorator for callback query handler607        Example:608        .. code-block:: python3609            @dp.callback_query_handler(func=lambda callback_query: True)610            async def some_callback_handler(callback_query: types.CallbackQuery)611        :param func: custom any callable object612        :param state:613        :param custom_filters:614        :param run_task: run callback in task (no wait results)615        :param kwargs:616        """617        def decorator(callback):618            self.register_callback_query_handler(callback, func=func, state=state, custom_filters=custom_filters,619                                                 run_task=run_task, **kwargs)620            return callback621        return decorator622    def register_shipping_query_handler(self, callback, *, func=None, state=None, custom_filters=None, run_task=None,623                                        **kwargs):624        """625        Register handler for shipping query626        Example:627        .. code-block:: python3628            dp.register_shipping_query_handler(some_shipping_query_handler, func=lambda shipping_query: True)629        :param callback:630        :param func: custom any callable object631        :param state:632        :param custom_filters:633        :param run_task: run callback in task (no wait results)634        :param kwargs:635        """636        if custom_filters is None:637            custom_filters = []638        filters_set = generate_default_filters(self,639                                               *custom_filters,640                                               func=func,641                                               state=state,642                                               **kwargs)643        self.shipping_query_handlers.register(self._wrap_async_task(callback, run_task), filters_set)644    def shipping_query_handler(self, *custom_filters, func=None, state=None, run_task=None, **kwargs):645        """646        Decorator for shipping query handler647        Example:648        .. code-block:: python3649            @dp.shipping_query_handler(func=lambda shipping_query: True)650            async def some_shipping_query_handler(shipping_query: types.ShippingQuery)651        :param func: custom any callable object652        :param state:653        :param custom_filters:654        :param run_task: run callback in task (no wait results)655        :param kwargs:656        """657        def decorator(callback):658            self.register_shipping_query_handler(callback, func=func, state=state, custom_filters=custom_filters,659                                                 run_task=run_task, **kwargs)660            return callback661        return decorator662    def register_pre_checkout_query_handler(self, callback, *, func=None, state=None, custom_filters=None,663                                            run_task=None, **kwargs):664        """665        Register handler for pre-checkout query666        Example:667        .. code-block:: python3668            dp.register_pre_checkout_query_handler(some_pre_checkout_query_handler, func=lambda shipping_query: True)669        :param callback:670        :param func: custom any callable object671        :param state:672        :param custom_filters:673        :param run_task: run callback in task (no wait results)674        :param kwargs:675        """676        if custom_filters is None:677            custom_filters = []678        filters_set = generate_default_filters(self,679                                               *custom_filters,680                                               func=func,681                                               state=state,682                                               **kwargs)683        self.pre_checkout_query_handlers.register(self._wrap_async_task(callback, run_task), filters_set)684    def pre_checkout_query_handler(self, *custom_filters, func=None, state=None, run_task=None, **kwargs):685        """686        Decorator for pre-checkout query handler687        Example:688        .. code-block:: python3689            @dp.pre_checkout_query_handler(func=lambda shipping_query: True)690            async def some_pre_checkout_query_handler(shipping_query: types.ShippingQuery)691        :param func: custom any callable object692        :param state:693        :param custom_filters:694        :param run_task: run callback in task (no wait results)695        :param kwargs:696        """697        def decorator(callback):698            self.register_pre_checkout_query_handler(callback, func=func, state=state, custom_filters=custom_filters,699                                                     run_task=run_task, **kwargs)700            return callback701        return decorator702    def register_errors_handler(self, callback, *, func=None, exception=None, run_task=None):703        """704        Register handler for errors705        :param callback:706        :param func:707        :param exception: you can make handler for specific errors type708        :param run_task: run callback in task (no wait results)709        """710        filters_set = []711        if func is not None:712            filters_set.append(func)713        if exception is not None:714            filters_set.append(ExceptionsFilter(exception))715        self.errors_handlers.register(self._wrap_async_task(callback, run_task), filters_set)716    def errors_handler(self, func=None, exception=None, run_task=None):717        """718        Decorator for errors handler719        :param func:720        :param exception: you can make handler for specific errors type721        :param run_task: run callback in task (no wait results)722        :return:723        """724        def decorator(callback):725            self.register_errors_handler(self._wrap_async_task(callback, run_task),726                                         func=func, exception=exception)727            return callback728        return decorator729    def current_state(self, *,730                      chat: typing.Union[str, int, None] = None,731                      user: typing.Union[str, int, None] = None) -> FSMContext:732        """733        Get current state for user in chat as context734        .. code-block:: python3735            with dp.current_state(chat=message.chat.id, user=message.user.id) as state:736                pass737            state = dp.current_state()738            state.set_state('my_state')739        :param chat:740        :param user:741        :return:742        """743        if chat is None:744            from .ctx import get_chat745            chat = get_chat()746        if user is None:747            from .ctx import get_user748            user = get_user()749        return FSMContext(storage=self.storage, chat=chat, user=user)750    async def throttle(self, key, *, rate=None, user=None, chat=None, no_error=None) -> bool:751        """752        Execute throttling manager.753        Returns True if limit has not exceeded otherwise raises ThrottleError or returns False754        :param key: key in storage755        :param rate: limit (by default is equal to default rate limit)756        :param user: user id757        :param chat: chat id758        :param no_error: return boolean value instead of raising error759        :return: bool760        """761        if not self.storage.has_bucket():762            raise RuntimeError('This storage does not provide Leaky Bucket')763        if no_error is None:764            no_error = self.no_throttle_error765        if rate is None:766            rate = self.throttling_rate_limit767        if user is None and chat is None:768            from . import ctx769            user = ctx.get_user()770            chat = ctx.get_chat()771        # Detect current time772        now = time.time()773        bucket = await self.storage.get_bucket(chat=chat, user=user)774        # Fix bucket775        if bucket is None:776            bucket = {key: {}}777        if key not in bucket:778            bucket[key] = {}779        data = bucket[key]780        # Calculate781        called = data.get(LAST_CALL, now)782        delta = now - called783        result = delta >= rate or delta <= 0784        # Save results785        data[RESULT] = result786        data[RATE_LIMIT] = rate787        data[LAST_CALL] = now788        data[DELTA] = delta789        if not result:790            data[EXCEEDED_COUNT] += 1791        else:792            data[EXCEEDED_COUNT] = 1793        bucket[key].update(data)794        await self.storage.set_bucket(chat=chat, user=user, bucket=bucket)795        if not result and not no_error:796            # Raise if it is allowed797            raise Throttled(key=key, chat=chat, user=user, **data)798        return result799    async def check_key(self, key, chat=None, user=None):800        """801        Get information about key in bucket802        :param key:803        :param chat:804        :param user:805        :return:806        """807        if not self.storage.has_bucket():808            raise RuntimeError('This storage does not provide Leaky Bucket')809        if user is None and chat is None:810            from . import ctx811            user = ctx.get_user()812            chat = ctx.get_chat()813        bucket = await self.storage.get_bucket(chat=chat, user=user)814        data = bucket.get(key, {})815        return Throttled(key=key, chat=chat, user=user, **data)816    async def release_key(self, key, chat=None, user=None):817        """818        Release blocked key819        :param key:820        :param chat:821        :param user:822        :return:823        """824        if not self.storage.has_bucket():825            raise RuntimeError('This storage does not provide Leaky Bucket')826        if user is None and chat is None:827            from . import ctx828            user = ctx.get_user()829            chat = ctx.get_chat()830        bucket = await self.storage.get_bucket(chat=chat, user=user)831        if bucket and key in bucket:832            del bucket['key']833            await self.storage.set_bucket(chat=chat, user=user, bucket=bucket)834            return True835        return False836    def async_task(self, func):837        """838        Execute handler as task and return None.839        Use this decorator for slow handlers (with timeouts)840        .. code-block:: python3841            @dp.message_handler(commands=['command'])842            @dp.async_task843            async def cmd_with_timeout(message: types.Message):844                await asyncio.sleep(120)845                return SendMessage(message.chat.id, 'KABOOM').reply(message)846        :param func:847        :return:848        """849        def process_response(task):850            try:851                response = task.result()852            except Exception as e:853                self.loop.create_task(854                    self.errors_handlers.notify(self, task.context.get(UPDATE_OBJECT, None), e))855            else:856                if isinstance(response, BaseResponse):857                    self.loop.create_task(response.execute_response(self.bot))858        @functools.wraps(func)859        async def wrapper(*args, **kwargs):860            task = self.loop.create_task(func(*args, **kwargs))861            task.add_done_callback(process_response)862        return wrapper863    def _wrap_async_task(self, callback, run_task=None) -> callable:864        if run_task is None:865            run_task = self.run_tasks_by_default866        if run_task:867            return self.async_task(callback)...

node.py

Source:node.py

...300                newest_file = max(abs_files, key=os.path.getctime)301                run_task['flags'].append(newest_file)302                # print run_task303                # print run_task['executable'], run_task['flags']304                self.main_window.run_task(node_object=self, executable=run_task['executable'], args=run_task['flags'])305                # run_task['flags'].remove(newest_file)306            else:307                ok, job_deadline = jobDeadlineUi.getDeadlineJobData(self.location, self.main_window)308                if ok:309                    txt_file = os.path.join(self.location, 'project', 'deadlineJob.txt')310                    job_file = open(txt_file, 'w')311                    for element in job_deadline:312                        job_file.write(element)313                        job_file.write(' ')314                    job_file.close()315                    self.main_window.submitDeadlineJob(txt_file)316    def data_ready(self):317        cursor_box = self.main_window.statusBox.textCursor()318        cursor_box.movePosition(cursor_box.End)...

test_rules.py

Source:test_rules.py

...21        q1 = ChessPieceFactory.create_piece('queen', 1, 1)22        q2 = ChessPieceFactory.create_piece('queen', 2, 3)23        assert not q1.collision(q2)24    def test_queen_board_size_1(self, mock_update_db):25        assert 1 == run_task(1, 1, 'queen', 'a', 'b')26    27    def test_queen_board_size_1(self, mock_update_db):28        assert 0 == run_task(2, 2, 'queen', 'a', 'b')29    def test_queen_board_size_1(self, mock_update_db):30        assert 0 == run_task(3, 3, 'queen', 'a', 'b')31    def test_queen_board_size_1(self, mock_update_db):32        assert 2 == run_task(4, 4, 'queen', 'a', 'b')33    def test_queen_board_size_1(self, mock_update_db):34        assert 10 == run_task(5, 5, 'queen', 'a', 'b')35    def test_queen_board_size_1(self, mock_update_db):36        assert 4 == run_task(6, 6, 'queen', 'a', 'b')37    def test_queen_board_size_1(self, mock_update_db):38        assert 40 == run_task(7, 7, 'queen', 'a', 'b')39    def test_queen_board_size_1(self, mock_update_db):40        assert 92 == run_task(8, 8, 'queen', 'a', 'b')41class Test_Rook():42    def test_rook_board_size_1(self, mock_update_db):43        assert 1 == run_task(1, 1, 'rook', 'a', 'b')44    def test_rook_board_size_2(self, mock_update_db):45        assert 2 == run_task(2, 2, 'rook', 'a', 'b')46    47    def test_rook_board_size_3(self, mock_update_db):48        assert 6 == run_task(3, 3, 'rook', 'a', 'b')49class Test_Bishop():50    def test_bishop_board_size_1(self, mock_update_db):51        assert 1 == run_task(1, 1, 'bishop', 'a', 'b')52    def test_bishop_board_size_2(self, mock_update_db):53        assert 4 == run_task(2, 2, 'bishop', 'a', 'b')54    55    def test_bishop_board_size_3(self, mock_update_db):56        assert 26 == run_task(3, 3, 'bishop', 'a', 'b')57class Test_Knight():58    def test_knight_board_size_1(self, mock_update_db):59        assert 1 == run_task(1, 1, 'knight', 'a', 'b')60    def test_knight_board_size_2(self, mock_update_db):61        assert 6 == run_task(2, 2, 'knight', 'a', 'b')62    63    def test_knight_board_size_3(self, mock_update_db):64        assert 36 == run_task(3, 3, 'knight', 'a', 'b')65class Test_King():66    def test_king_board_size_1(self, mock_update_db):67        assert 1 == run_task(1, 1, 'king', 'a', 'b')68    def test_king_board_size_2(self, mock_update_db):69        assert 0 == run_task(2, 2, 'king', 'a', 'b')70    def test_king_board_size_3(self, mock_update_db):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.