Dispatcher Module

Dispatcher(router_id=None, storage=MemoryContext, *, use_create_task=False, **storage_kwargs)

Bases: BotMixin

Основной класс для обработки событий бота.

Обеспечивает запуск поллинга и вебхука, маршрутизацию событий, применение middleware, фильтров и вызов соответствующих обработчиков.

Инициализация диспетчера.

Parameters:

Name	Type	Description	Default
router_id	str | None	Идентификатор роутера для логов.	None
use_create_task	bool	Флаг, отвечающий за параллелизацию обработок событий.	False
storage	type[BaseContext]	Класс контекста для хранения данных (MemoryContext, RedisContext и т.д.).	MemoryContext
**storage_kwargs	Any	Дополнительные аргументы для инициализации хранилища.	{}

Source code in maxapi/dispatcher.py

def __init__(
    self,
    router_id: str | None = None,
    storage: Any = MemoryContext,
    *,
    use_create_task: bool = False,
    **storage_kwargs: Any,
) -> None:
    """
    Инициализация диспетчера.

    Args:
        router_id (str | None): Идентификатор роутера для логов.
        use_create_task (bool): Флаг, отвечающий за параллелизацию
            обработок событий.
        storage (type[BaseContext]): Класс контекста для хранения
            данных (MemoryContext, RedisContext и т.д.).
        **storage_kwargs (Any): Дополнительные аргументы для
            инициализации хранилища.
    """

    self.router_id = router_id
    self.storage = storage
    self.storage_kwargs = storage_kwargs

    self.event_handlers: list[Handler] = []
    self.contexts: dict[tuple[int | None, int | None], BaseContext] = {}
    self.routers: list[Router | Dispatcher] = []
    self.filters: list[MagicFilter] = []
    self.base_filters: list[BaseFilter] = []
    self.middlewares: list[BaseMiddleware] = []

    self.bot: Bot | None = None
    self.on_started_func: Callable | None = None
    self.polling = False
    self.use_create_task = use_create_task

    self.message_created = Event(
        update_type=UpdateType.MESSAGE_CREATED, router=self
    )
    self.bot_added = Event(update_type=UpdateType.BOT_ADDED, router=self)
    self.bot_removed = Event(
        update_type=UpdateType.BOT_REMOVED, router=self
    )
    self.bot_started = Event(
        update_type=UpdateType.BOT_STARTED, router=self
    )
    self.bot_stopped = Event(
        update_type=UpdateType.BOT_STOPPED, router=self
    )
    self.dialog_cleared = Event(
        update_type=UpdateType.DIALOG_CLEARED, router=self
    )
    self.dialog_muted = Event(
        update_type=UpdateType.DIALOG_MUTED, router=self
    )
    self.dialog_unmuted = Event(
        update_type=UpdateType.DIALOG_UNMUTED, router=self
    )
    self.dialog_removed = Event(
        update_type=UpdateType.DIALOG_REMOVED, router=self
    )
    self.raw_api_response = Event(
        update_type=UpdateType.RAW_API_RESPONSE, router=self
    )
    self.chat_title_changed = Event(
        update_type=UpdateType.CHAT_TITLE_CHANGED, router=self
    )
    self.message_callback = Event(
        update_type=UpdateType.MESSAGE_CALLBACK, router=self
    )
    self.message_chat_created = Event(
        update_type=UpdateType.MESSAGE_CHAT_CREATED,
        router=self,
        deprecated=True,
    )
    self.message_edited = Event(
        update_type=UpdateType.MESSAGE_EDITED, router=self
    )
    self.message_removed = Event(
        update_type=UpdateType.MESSAGE_REMOVED, router=self
    )
    self.user_added = Event(update_type=UpdateType.USER_ADDED, router=self)
    self.user_removed = Event(
        update_type=UpdateType.USER_REMOVED, router=self
    )
    self.on_started = Event(update_type=UpdateType.ON_STARTED, router=self)

check_me() async

Проверяет и логирует информацию о боте.

Source code in maxapi/dispatcher.py

async def check_me(self) -> None:
    """
    Проверяет и логирует информацию о боте.
    """

    me = await self._ensure_bot().get_me()

    self._ensure_bot()._me = me  # noqa: SLF001

    logger_dp.info(
        f"Бот: @{me.username} first_name={me.first_name} id={me.user_id}"
    )

build_middleware_chain(middlewares, handler)

Формирует цепочку вызова middleware вокруг хендлера.

Parameters:

Name	Type	Description	Default
middlewares	List[BaseMiddleware]	Список middleware.	required
handler	Callable	Финальный обработчик.	required

Returns:

Name	Type	Description
Callable	Callable[[Any, dict[str, Any]], Awaitable[Any]]	Обёрнутый обработчик.

Source code in maxapi/dispatcher.py

def build_middleware_chain(
    self,
    middlewares: list[BaseMiddleware],
    handler: Callable[[Any, dict[str, Any]], Awaitable[Any]],
) -> Callable[[Any, dict[str, Any]], Awaitable[Any]]:
    """
    Формирует цепочку вызова middleware вокруг хендлера.

    Args:
        middlewares (List[BaseMiddleware]): Список middleware.
        handler (Callable): Финальный обработчик.

    Returns:
        Callable: Обёрнутый обработчик.
    """

    for mw in reversed(middlewares):
        handler = functools.partial(mw, handler)

    return handler

include_routers(*routers)

Добавляет указанные роутеры в диспетчер.

Parameters:

Name	Type	Description	Default
*routers	Router	Роутеры для добавления.	()

Source code in maxapi/dispatcher.py

def include_routers(self, *routers: Router) -> None:
    """
    Добавляет указанные роутеры в диспетчер.

    Args:
        *routers (Router): Роутеры для добавления.
    """

    self.routers.extend(routers)

outer_middleware(middleware)

Добавляет Middleware на первое место в списке.

Parameters:

Name	Type	Description	Default
middleware	BaseMiddleware	Middleware.	required

Source code in maxapi/dispatcher.py

def outer_middleware(self, middleware: BaseMiddleware) -> None:
    """
    Добавляет Middleware на первое место в списке.

    Args:
        middleware (BaseMiddleware): Middleware.
    """

    self.middlewares.insert(0, middleware)

middleware(middleware)

Добавляет Middleware в конец списка.

Parameters:

Name	Type	Description	Default
middleware	BaseMiddleware	Middleware.	required

Source code in maxapi/dispatcher.py

def middleware(self, middleware: BaseMiddleware) -> None:
    """
    Добавляет Middleware в конец списка.

    Args:
        middleware (BaseMiddleware): Middleware.
    """

    self.middlewares.append(middleware)

filter(base_filter)

Добавляет фильтр в список.

Parameters:

Name	Type	Description	Default
base_filter	BaseFilter	Фильтр.	required

Source code in maxapi/dispatcher.py

def filter(self, base_filter: BaseFilter) -> None:
    """
    Добавляет фильтр в список.

    Args:
        base_filter (BaseFilter): Фильтр.
    """

    self.base_filters.append(base_filter)

__ready(bot) async

Подготавливает диспетчер: сохраняет бота, подготавливает обработчики, вызывает on_started.

Parameters:

Name	Type	Description	Default
bot	Bot	Экземпляр бота.	required

Source code in maxapi/dispatcher.py

async def __ready(self, bot: Bot) -> None:
    """
    Подготавливает диспетчер: сохраняет бота, подготавливает
    обработчики, вызывает on_started.

    Args:
        bot (Bot): Экземпляр бота.
    """

    self.bot = bot
    self.bot.dispatcher = self

    if self.polling and bot.auto_check_subscriptions:
        await self._check_subscriptions(bot)

    await self.check_me()

    self.routers += [self]
    self._prepare_handlers(bot)

    if self.on_started_func:
        await self.on_started_func()

__get_context(chat_id, user_id)

Возвращает существующий или создаёт новый контекст по chat_id и user_id.

Parameters:

Name	Type	Description	Default
chat_id	Optional[int]	Идентификатор чата.	required
user_id	Optional[int]	Идентификатор пользователя.	required

Returns:

Name	Type	Description
BaseContext	BaseContext	Контекст.

Source code in maxapi/dispatcher.py

def __get_context(
    self, chat_id: int | None, user_id: int | None
) -> BaseContext:
    """
    Возвращает существующий или создаёт новый контекст
    по chat_id и user_id.

    Args:
        chat_id (Optional[int]): Идентификатор чата.
        user_id (Optional[int]): Идентификатор пользователя.

    Returns:
        BaseContext: Контекст.
    """

    key = (chat_id, user_id)
    if key in self.contexts:
        return self.contexts[key]

    new_ctx = self.storage(chat_id, user_id, **self.storage_kwargs)
    self.contexts[key] = new_ctx
    return new_ctx

call_handler(handler, event_object, data) async

Вызывает хендлер с нужными аргументами.

Parameters:

Name	Type	Description	Default
handler	Handler	Handler.	required
event_object	UpdateType | dict[str, Any]	Объект события.	required
data	dict[str, Any]	Данные для хендлера.	required

Returns:

Type	Description
None	None

Source code in maxapi/dispatcher.py

async def call_handler(
    self,
    handler: Handler,
    event_object: UpdateType | dict[str, Any],
    data: dict[str, Any],
) -> None:
    """
    Вызывает хендлер с нужными аргументами.

    Args:
        handler: Handler.
        event_object: Объект события.
        data: Данные для хендлера.

    Returns:
        None
    """

    func_args = handler.func_event.__annotations__.keys()
    kwargs_filtered = {k: v for k, v in data.items() if k in func_args}

    if kwargs_filtered:
        await handler.func_event(event_object, **kwargs_filtered)
    else:
        await handler.func_event(event_object)

process_base_filters(event, filters) async

Асинхронно применяет фильтры к событию.

Parameters:

Name	Type	Description	Default
event	UpdateUnion	Событие.	required
filters	List[BaseFilter]	Список фильтров.	required

Returns:

Type	Description
dict[str, Any] | None | Literal[False]	Optional[Dict[str, Any]] | Literal[False]: Словарь с результатом или False.

Source code in maxapi/dispatcher.py

async def process_base_filters(
    self, event: UpdateUnion, filters: list[BaseFilter]
) -> dict[str, Any] | None | Literal[False]:
    """
    Асинхронно применяет фильтры к событию.

    Args:
        event (UpdateUnion): Событие.
        filters (List[BaseFilter]): Список фильтров.

    Returns:
        Optional[Dict[str, Any]] | Literal[False]: Словарь с
            результатом или False.
    """

    data = {}

    for _filter in filters:
        result = await _filter(event)

        if isinstance(result, dict):
            data.update(result)

        elif not result:
            return result

    return data

handle_raw_response(event_type, raw_data) async

Специальный метод для обработки сырых ответов API.

Source code in maxapi/dispatcher.py

async def handle_raw_response(
    self, event_type: UpdateType, raw_data: dict[str, Any]
) -> None:
    """
    Специальный метод для обработки сырых ответов API.
    """
    for router, *_ in self._iter_unique_routers(self.routers):
        matching_handlers = self._find_matching_handlers(
            router, event_type
        )
        for handler in matching_handlers:
            try:
                await self.call_handler(handler, raw_data, {})
            except Exception as e:  # noqa: PERF203
                logger_dp.exception(
                    f"Ошибка в обработчике RAW_API_RESPONSE: {e}"
                )

handle(event_object) async

Основной обработчик события. Применяет фильтры, middleware и вызывает нужный handler.

Parameters:

Name	Type	Description	Default
event_object	UpdateUnion	Событие.	required

Source code in maxapi/dispatcher.py

async def handle(self, event_object: UpdateUnion) -> None:
    """
    Основной обработчик события. Применяет фильтры, middleware
    и вызывает нужный handler.

    Args:
        event_object (UpdateUnion): Событие.
    """
    router_id = None
    process_info = "нет данных"

    try:
        ids = event_object.get_ids()
        memory_context = self.__get_context(*ids)
        current_state = await memory_context.get_state()
        kwargs = {"context": memory_context}

        process_info = (
            f"{event_object.update_type} | "
            f"chat_id: {ids[0]}, user_id: {ids[1]}"
        )

        is_handled = False

        async def _process_event(
            _: UpdateUnion, data: dict[str, Any]
        ) -> None:
            nonlocal router_id, is_handled, memory_context, current_state

            data["context"] = memory_context

            for index, (
                router,
                router_middlewares,
                router_filters,
                router_base_filters,
            ) in enumerate(self._iter_unique_routers(self.routers)):
                if is_handled:
                    break

                router_id = router.router_id or index

                router_filter_result = await self._check_router_filters(
                    event_object, router_filters, router_base_filters
                )

                if router_filter_result is False:
                    continue

                if isinstance(router_filter_result, dict):
                    data.update(router_filter_result)

                matching_handlers = self._find_matching_handlers(
                    router, event_object.update_type
                )

                if not matching_handlers:
                    continue

                async def _process_handlers(
                    event: UpdateUnion, handler_data: dict[str, Any]
                ) -> None:
                    nonlocal is_handled

                    for handler in matching_handlers:
                        handler_match_result = (
                            await self._check_handler_match(
                                handler, event, current_state
                            )
                        )

                        if handler_match_result is False:
                            continue

                        if isinstance(handler_match_result, dict):
                            handler_data.update(handler_match_result)

                        await self._execute_handler(
                            handler=handler,
                            event=event,
                            data=handler_data,
                            handler_middlewares=handler.middlewares,
                            memory_context=memory_context,
                            current_state=current_state,
                            router_id=router_id,
                            process_info=process_info,
                        )

                        logger_dp.info(
                            f"Обработано: "
                            f"router_id: {router_id} | {process_info}"
                        )

                        is_handled = True
                        break

                if router_middlewares:
                    router_chain = self.build_middleware_chain(
                        router_middlewares, _process_handlers
                    )
                    await router_chain(event_object, data)
                else:
                    await _process_handlers(event_object, data)

        global_chain = self.build_middleware_chain(
            self.middlewares, _process_event
        )

        try:
            await global_chain(event_object, kwargs)
        except Exception as e:
            mem_data = await memory_context.get_data()

            if hasattr(global_chain, "func"):
                middleware_title = global_chain.func.__class__.__name__
            else:
                middleware_title = getattr(
                    global_chain,
                    "__name__",
                    global_chain.__class__.__name__,
                )

            raise MiddlewareException(
                middleware_title=middleware_title,
                router_id=router_id,
                process_info=process_info,
                memory_context={
                    "data": mem_data,
                    "state": current_state,
                },
                cause=e,
            ) from e

        if not is_handled:
            logger_dp.info(
                f"Проигнорировано: router_id: {router_id} | {process_info}"
            )

    except Exception as e:
        logger_dp.exception(
            f"Ошибка при обработке события: router_id: "
            f"{router_id} | {process_info} | {e} "
        )

start_polling(bot, *, skip_updates=False) async

Запускает цикл получения обновлений (long polling).

Parameters:

Name	Type	Description	Default
bot	Bot	Экземпляр бота.	required
skip_updates	bool	Флаг, отвечающий за обработку старых событий.	False

Source code in maxapi/dispatcher.py

async def start_polling(
    self, bot: Bot, *, skip_updates: bool = False
) -> None:
    """
    Запускает цикл получения обновлений (long polling).

    Args:
        bot (Bot): Экземпляр бота.
        skip_updates (bool): Флаг, отвечающий за обработку старых событий.
    """

    self.polling = True

    await self.__ready(bot)

    current_timestamp = to_ms(datetime.now())

    while self.polling:
        try:
            events: dict = await self._ensure_bot().get_updates(
                marker=self._ensure_bot().marker_updates
            )
        except AsyncioTimeoutError:
            continue
        except (MaxConnection, ClientConnectorError) as e:
            logger_dp.warning(
                f"Ошибка подключения при получении обновлений: {e}, "
                f"жду {CONNECTION_RETRY_DELAY} секунд"
            )
            await asyncio.sleep(CONNECTION_RETRY_DELAY)
            continue
        except InvalidToken:
            logger_dp.error("Неверный токен! Останавливаю polling")
            self.polling = False
            raise
        except MaxApiError as e:
            logger_dp.info(
                f"Ошибка при получении обновлений: {e}, "
                f"жду {GET_UPDATES_RETRY_DELAY} секунд"
            )
            await asyncio.sleep(GET_UPDATES_RETRY_DELAY)
            continue
        except Exception as e:
            logger_dp.error(
                f"Неожиданная ошибка при получении обновлений: "
                f"{e.__class__.__name__}: {e}"
            )
            await asyncio.sleep(GET_UPDATES_RETRY_DELAY)
            continue

        try:
            self._ensure_bot().marker_updates = events.get("marker")

            processed_events = await process_update_request(
                events=events, bot=self._ensure_bot()
            )

            for event in processed_events:
                if skip_updates and event.timestamp < current_timestamp:
                    logger_dp.info(
                        f"Пропуск события от {from_ms(event.timestamp)}: "
                        f"{event.update_type}"
                    )
                    continue

                if self.use_create_task:
                    asyncio.create_task(self.handle(event))

                else:
                    await self.handle(event)

        except ClientConnectorError:
            logger_dp.error(
                f"Ошибка подключения, жду {CONNECTION_RETRY_DELAY} секунд"
            )
            await asyncio.sleep(CONNECTION_RETRY_DELAY)
        except Exception as e:
            logger_dp.error(
                f"Общая ошибка при обработке событий: {e.__class__} - {e}"
            )

stop_polling() async

Останавливает цикл получения обновлений (long polling).

Этот метод устанавливает флаг polling в False, что приводит к завершению цикла в методе start_polling.

Source code in maxapi/dispatcher.py

async def stop_polling(self) -> None:
    """
    Останавливает цикл получения обновлений (long polling).

    Этот метод устанавливает флаг polling в False, что приводит к
    завершению цикла в методе start_polling.
    """
    if self.polling:
        self.polling = False
        logger_dp.info("Polling остановлен")

startup(bot) async

Инициализирует диспетчер: сохраняет бота, подготавливает обработчики и вызывает on_started.

Используется интеграционными модулями (например, maxapi.webhook.fastapi) для инициализации в lifespan веб-фреймворка.

Parameters:

Name	Type	Description	Default
bot	Bot	Экземпляр бота.	required

Source code in maxapi/dispatcher.py

async def startup(self, bot: Bot) -> None:
    """
    Инициализирует диспетчер: сохраняет бота, подготавливает
    обработчики и вызывает on_started.

    Используется интеграционными модулями (например,
    maxapi.webhook.fastapi) для инициализации в lifespan
    веб-фреймворка.

    Args:
        bot (Bot): Экземпляр бота.
    """
    await self.__ready(bot)

handle_webhook(bot, *, host=DEFAULT_HOST, port=DEFAULT_PORT, path=DEFAULT_PATH, secret=None, webhook_type=AiohttpMaxWebhook, **kwargs) async

Запускает вебхук-сервер (aiohttp) для приёма обновлений.

Удобный метод «всё в одном»: создаёт aiohttp-приложение через :class:~maxapi.webhook.aiohttp.BaseMaxWebhook, регистрирует маршрут и запускает сервер.

Для более гибкого управления жизненным циклом сервера используйте одну из реализаций BaseMaxWebhook напрямую, например :class:~maxapi.webhook.aiohttp.BaseMaxWebhook.

Parameters:

Name	Type	Description	Default
bot	Bot	Экземпляр бота.	required
host	str	Хост сервера (по умолчанию "0.0.0.0").	DEFAULT_HOST
port	int	Порт сервера (по умолчанию 8080).	DEFAULT_PORT
path	str	URL-путь для маршрута вебхука.	DEFAULT_PATH
secret	str | None	Секрет для проверки заголовка X-Max-Bot-Api-Secret. Должен совпадать со значением, переданным в :meth:~maxapi.Bot.subscribe_webhook.	None
webhook_type	type[BaseMaxWebhook]	Класс вебхука.	AiohttpMaxWebhook
**kwargs	Any	Дополнительные аргументы для aiohttp.web.AppRunner.	{}

Source code in maxapi/dispatcher.py

async def handle_webhook(
    self,
    bot: Bot,
    *,
    host: str = DEFAULT_HOST,
    port: int = DEFAULT_PORT,
    path: str = DEFAULT_PATH,
    secret: str | None = None,
    webhook_type: type[BaseMaxWebhook] = AiohttpMaxWebhook,
    **kwargs: Any,
) -> None:
    """
    Запускает вебхук-сервер (aiohttp) для приёма обновлений.

    Удобный метод «всё в одном»: создаёт aiohttp-приложение через
    :class:`~maxapi.webhook.aiohttp.BaseMaxWebhook`,
    регистрирует маршрут и запускает сервер.

    Для более гибкого управления жизненным циклом сервера используйте
    одну из реализаций BaseMaxWebhook напрямую, например
    :class:`~maxapi.webhook.aiohttp.BaseMaxWebhook`.

    Args:
        bot (Bot): Экземпляр бота.
        host (str): Хост сервера (по умолчанию ``"0.0.0.0"``).
        port (int): Порт сервера (по умолчанию ``8080``).
        path (str): URL-путь для маршрута вебхука.
        secret (str | None): Секрет для проверки заголовка
            ``X-Max-Bot-Api-Secret``. Должен совпадать со значением,
            переданным в :meth:`~maxapi.Bot.subscribe_webhook`.
        webhook_type (type[BaseMaxWebhook]): Класс вебхука.
        **kwargs: Дополнительные аргументы для ``aiohttp.web.AppRunner``.
    """
    webhook = webhook_type(dp=self, bot=bot, secret=secret)
    await webhook.run(host=host, port=port, path=path, **kwargs)

init_serve(bot, host=DEFAULT_HOST, port=DEFAULT_PORT, **kwargs) async

.. deprecated:: Используйте :meth:handle_webhook вместо init_serve. Метод будет удалён в одной из следующих версий.

Parameters:

Name	Type	Description	Default
bot	Bot	Экземпляр бота.	required
host	str	Хост.	DEFAULT_HOST
port	int	Порт.	DEFAULT_PORT

Source code in maxapi/dispatcher.py

async def init_serve(  # pragma: no cover
    self,
    bot: Bot,
    host: str = DEFAULT_HOST,
    port: int = DEFAULT_PORT,
    **kwargs: Any,
) -> None:
    """
    .. deprecated::
        Используйте :meth:`handle_webhook` вместо ``init_serve``.
        Метод будет удалён в одной из следующих версий.

    Args:
        bot (Bot): Экземпляр бота.
        host (str): Хост.
        port (int): Порт.
    """
    warn(
        "init_serve устарел и будет удалён в следующих версиях. "
        "Используйте handle_webhook вместо него.",
        DeprecationWarning,
        stacklevel=2,
    )
    await self.handle_webhook(bot, host=host, port=port, **kwargs)

Router(router_id=None)

Bases: Dispatcher

Роутер для группировки обработчиков событий.

Инициализация роутера.

Parameters:

Name	Type	Description	Default
router_id	str | None	Идентификатор роутера для логов.	None

Source code in maxapi/dispatcher.py

def __init__(self, router_id: str | None = None):
    """
    Инициализация роутера.

    Args:
        router_id (str | None): Идентификатор роутера для логов.
    """

    super().__init__(router_id)

Event(update_type, router, *, deprecated=False)

Декоратор для регистрации обработчиков событий.

Инициализирует событие-декоратор.

Parameters:

Name	Type	Description	Default
update_type	UpdateType	Тип события.	required
router	Dispatcher | Router	Экземпляр роутера или диспетчера.	required
deprecated	bool	Флаг, указывающий на то, что событие устарело.	False

Source code in maxapi/dispatcher.py

def __init__(
    self,
    update_type: UpdateType,
    router: Dispatcher | Router,
    *,
    deprecated: bool = False,
):
    """
    Инициализирует событие-декоратор.

    Args:
        update_type (UpdateType): Тип события.
        router (Dispatcher | Router): Экземпляр роутера или диспетчера.
        deprecated (bool): Флаг, указывающий на то, что событие устарело.
    """

    self.update_type = update_type
    self.router = router
    self.deprecated = deprecated

register(func_event, *args, **kwargs)

Регистрирует функцию как обработчик события.

Parameters:

Name	Type	Description	Default
func_event	Callable	Функция-обработчик	required
*args	Any	Фильтры	()
**kwargs	Any	Дополнительные параметры (например, states)	{}

Returns:

Name	Type	Description
Callable	Callable	Исходная функция.

Source code in maxapi/dispatcher.py

def register(
    self, func_event: Callable, *args: Any, **kwargs: Any
) -> Callable:
    """
    Регистрирует функцию как обработчик события.

    Args:
        func_event (Callable): Функция-обработчик
        *args: Фильтры
        **kwargs: Дополнительные параметры (например, states)

    Returns:
        Callable: Исходная функция.
    """

    if self.deprecated:
        warnings.warn(
            f"Событие {self.update_type} устарело "
            f"и будет удалено в будущих версиях.",
            DeprecationWarning,
            stacklevel=3,
        )

    if self.update_type == UpdateType.ON_STARTED:
        self.router.on_started_func = func_event

    else:
        self.router.event_handlers.append(
            Handler(
                *args,
                func_event=func_event,
                update_type=self.update_type,
                **kwargs,
            )
        )
    return func_event

__call__(*args, **kwargs)

Регистрирует функцию как обработчик события через декоратор.

Returns:

Name	Type	Description
Callable	Callable	Декоратор.

Source code in maxapi/dispatcher.py

def __call__(self, *args: Any, **kwargs: Any) -> Callable:
    """
    Регистрирует функцию как обработчик события через декоратор.

    Returns:
        Callable: Декоратор.
    """

    def decorator(func_event: Callable) -> Callable:
        return self.register(func_event, *args, **kwargs)

    return decorator