Dispatcher and Router

Bases: Router

Top-level router that starts update polling.

Dispatcher extends Router and is intended to be the root object that receives updates from MaxApi and dispatches them to handlers.

Initialize the dispatcher.

Parameters:

Name	Type	Description	Default
storage	BaseStorage | None	BaseStorage instance to process.	None
fsm_strategy	FSMStrategy	FSMStrategy instance to process.	USER_IN_CHAT
events_isolation	BaseEventIsolation | None	BaseEventIsolation instance to process.	None
disable_fsm	bool	The disable fsm value.	False
name	str | None	The name value.	None
kwargs	Any	Keyword arguments forwarded to the wrapped callable.	{}

Source code in src/pyromax/dispatcher/Dispatcher.py

def __init__(
    self,
    *,
    storage: BaseStorage | None = None,
    fsm_strategy: FSMStrategy = FSMStrategy.USER_IN_CHAT,
    events_isolation: BaseEventIsolation | None = None,
    disable_fsm: bool = False,
    name: str | None = None,
    **kwargs: Any,
) -> None:
    """Initialize the dispatcher.

    :param storage: BaseStorage instance to process.
    :type storage: BaseStorage | None
    :param fsm_strategy: FSMStrategy instance to process.
    :type fsm_strategy: FSMStrategy
    :param events_isolation: BaseEventIsolation instance to process.
    :type events_isolation: BaseEventIsolation | None
    :param disable_fsm: The disable fsm value.
    :type disable_fsm: bool
    :param name: The name value.
    :type name: str | None
    :param kwargs: Keyword arguments forwarded to the wrapped callable.
    :type kwargs: Any
    """
    super().__init__(name=name)

    self.update = UpdateMaxEventObserver(
        router=self, event_name="UPDATE", type_of_update=MaxObject
    )

    async def notify_wrapper(
        resolved_update: ResolvedUpdate, data: DataDict
    ) -> Any:
        """Notify wrapper.

        :param resolved_update: ResolvedUpdate instance to process.
        :type resolved_update: ResolvedUpdate
        :param data: Contextual data passed through the processing pipeline.
        :type data: DataDict
        :returns: The value returned by the wrapped callable or backend.
        :rtype: Any
        """
        data.update(
            {
                type(resolved_update): resolved_update,
            }
        )
        result = await self.notify(resolved_update, data)
        if result is UNKNOWN_UPDATE:
            skip()
        return result

    self.update.register(notify_wrapper)

    self.update.outer_middleware(ErrorsMiddleware(self))

    self.update.outer_middleware(UserContextMiddleware())

    self.fsm = FSMContextMiddleware(
        storage=storage or MemoryStorage(),
        strategy=fsm_strategy,
        events_isolation=events_isolation or DisabledEventIsolation(),
    )

    if not disable_fsm:
        self.update.outer_middleware(self.fsm)

    self.__logger = logging.getLogger("MaxDispatcher")

update instance-attribute

update = UpdateMaxEventObserver(
    router=self,
    event_name="UPDATE",
    type_of_update=MaxObject,
)

fsm instance-attribute

fsm = FSMContextMiddleware(
    storage=storage or MemoryStorage(),
    strategy=fsm_strategy,
    events_isolation=events_isolation
    or DisabledEventIsolation(),
)

__logger instance-attribute

__logger = getLogger('MaxDispatcher')

start_polling async

start_polling(max_api: MaxApi) -> None

Start reading updates and dispatch them to handlers.

Parameters:

Name	Type	Description	Default
max_api	MaxApi	Initialized MaxApi instance.	required

Source code in src/pyromax/dispatcher/Dispatcher.py

async def start_polling(self, max_api: MaxApi) -> None:
    """Start reading updates and dispatch them to handlers.

    :param max_api: Initialized MaxApi instance.
    :type max_api: MaxApi
    """

    context = {"max_api": max_api}

    update_translator, updates = max_api.listen_updates(context=context)
    try:
        async for update in updates:

            self.__logger.debug("Received update: %s", update)

            resolved_update = update_translator(update)

            data: dict[type | TypeVar, Any] = {
                type(max_api): max_api,
                Update: update,
                ResolvedUpdate: resolved_update,
            }

            data.update(max_api.workflow_data)

            update_observer = self.update

            data[DataDict] = data

            response = await update_observer.wrap_outer_middleware(
                update_observer.update, update, data=data
            )

            handled = response is not UNHANDLED and response is not UNKNOWN_UPDATE

            self.__logger.debug(
                f'update %s was{"" if handled else "n`t"} handled: %s',
                update,
                handled,
            )
    finally:
        await self.fsm.close()

Bases: Subject

Container for handlers and nested routers.

Routers group event handlers into reusable modules and allow hierarchical composition of bot logic.

Attributes: events: a dict with all event observers(listeners)

Initialize the router.

Parameters:

Name	Type	Description	Default
name	str | None	The name value.	None

Source code in src/pyromax/dispatcher/Router.py

def __init__(
    self,
    *,
    name: str | None = None,
) -> None:
    """Initialize the router.

    :param name: The name value.
    :type name: str | None
    """
    self.name = name or hex(id(self))

    self.sub_routers: list[Router] = []
    self._parent_router: None | Router = None

    self.message = MessageEventObserver(self, "USER", type_of_update=Message)
    self.message_removed = RemovedMessageEventObserver(
        self, "REMOVED", type_of_update=Message
    )
    self.edited_message = MessageEventObserver(
        self, "EDITED", type_of_update=Message
    )
    self.reply_to_message = ReplyToMessageEventObserver(
        self, "REPLY", type_of_update=Message
    )
    self.forward_message = MessageForwardEventObserver(
        self, "FORWARD", type_of_update=Message
    )
    self.message_reaction = StandardMaxEventObserver(
        self, "MESSAGE_REACTION", type_of_update=EmojiReaction
    )
    self.message_added_reaction = EmojiReactionAddObserver(
        self, "MESSAGE_ADDED_REACTION", type_of_update=EmojiReaction
    )
    self.message_deleted_reaction = EmojiReactionRemoveObserver(
        self, "MESSAGE_DELETED_REACTION", type_of_update=EmojiReaction
    )
    self.error = StandardMaxEventObserver(self, "ERROR", type_of_update=ErrorEvent)
    # self.raw_update = UpdateMaxEventObserver(self, 'RAW_UPDATE', type_of_update=Response)
    self.events: dict[str, StandardMaxEventObserver[Any]] = {
        "EDITED": self.edited_message,
        "REPLY": self.reply_to_message,
        "FORWARD": self.forward_message,
        "REMOVED": self.message_removed,
        "USER": self.message,
        "MESSAGE_ADDED_REACTION": self.message_added_reaction,
        "MESSAGE_DELETED_REACTION": self.message_deleted_reaction,
        "MESSAGE_REACTION": self.message_reaction,
        "ERROR": self.error,
        # 'RAW_UPDATE': self.raw_update,
    }

name instance-attribute

name = name or hex(id(self))

sub_routers instance-attribute

sub_routers: list[Router] = []

_parent_router instance-attribute

_parent_router: None | Router = None

message instance-attribute

message = MessageEventObserver(
    self, "USER", type_of_update=Message
)

message_removed instance-attribute

message_removed = RemovedMessageEventObserver(
    self, "REMOVED", type_of_update=Message
)

edited_message instance-attribute

edited_message = MessageEventObserver(
    self, "EDITED", type_of_update=Message
)

reply_to_message instance-attribute

reply_to_message = ReplyToMessageEventObserver(
    self, "REPLY", type_of_update=Message
)

forward_message instance-attribute

forward_message = MessageForwardEventObserver(
    self, "FORWARD", type_of_update=Message
)

message_reaction instance-attribute

message_reaction = StandardMaxEventObserver(
    self, "MESSAGE_REACTION", type_of_update=EmojiReaction
)

message_added_reaction instance-attribute

message_added_reaction = EmojiReactionAddObserver(
    self,
    "MESSAGE_ADDED_REACTION",
    type_of_update=EmojiReaction,
)

message_deleted_reaction instance-attribute

message_deleted_reaction = EmojiReactionRemoveObserver(
    self,
    "MESSAGE_DELETED_REACTION",
    type_of_update=EmojiReaction,
)

error instance-attribute

error = StandardMaxEventObserver(
    self, "ERROR", type_of_update=ErrorEvent
)

events instance-attribute

events: dict[str, StandardMaxEventObserver[Any]] = {
    "EDITED": edited_message,
    "REPLY": reply_to_message,
    "FORWARD": forward_message,
    "REMOVED": message_removed,
    "USER": message,
    "MESSAGE_ADDED_REACTION": message_added_reaction,
    "MESSAGE_DELETED_REACTION": message_deleted_reaction,
    "MESSAGE_REACTION": message_reaction,
    "ERROR": error,
}

chain_head property

chain_head: Generator[Router, None, None]

Chain head.

:yields: Items produced by the iterator. :ytype: Generator['Router', None, None]

chain_tail property

chain_tail: Generator[Router, None, None]

Chain tail.

:yields: Items produced by the iterator. :ytype: Generator['Router', None, None]

parent_router property writable

parent_router: Optional[Router]

Parent router.

Returns:

Type	Description
Optional['Router']	The resulting Optional['Router'] value.

include_routers

include_routers(*routers: Router) -> None

Attach multiple child routers at once.

Parameters:

Name	Type	Description	Default
routers	Router	Routers to attach.	()

Raises:

Type	Description
ValueError	If the requested action cannot be completed.

Source code in src/pyromax/dispatcher/Router.py

def include_routers(self, *routers: "Router") -> None:
    """Attach multiple child routers at once.

    :param routers: Routers to attach.
    :type routers: 'Router'

    :raises ValueError: If the requested action cannot be completed.
    """
    if not routers:
        msg = "At least one router must be provided"
        raise ValueError(msg)
    for router in routers:
        self.include_router(router)

include_router

include_router(router: Router) -> Router

Attach another router as a child router.

Parameters:

Name	Type	Description	Default
router	Router	Router to attach.	required

Returns:

Type	Description
'Router'	The attached router.

Raises:

Type	Description
ValueError	If the requested action cannot be completed.

Source code in src/pyromax/dispatcher/Router.py

def include_router(self, router: "Router") -> "Router":
    """Attach another router as a child router.

    :param router: Router to attach.
    :type router: 'Router'

    :returns: The attached router.
    :rtype: 'Router'

    :raises ValueError: If the requested action cannot be completed.
    """
    if not isinstance(router, Router):
        msg = f"router should be instance of Router not {type(router).__class__.__name__}"
        raise ValueError(msg)
    router.parent_router = self
    return router

notify async

notify(
    update: MaxObject,
    data: dict[Any, Any] | None = None,
    event_types: list[str] | None = None,
) -> Any

Propagate an update through handlers and child routers.

Parameters:

Name	Type	Description	Default
update	MaxObject	Incoming update object.	required
data	dict[Any, Any] | None	Context data available to handlers.	None
event_types	list[str] | None	keys of Router.events	None

Returns:

Type	Description
Any	Any if the update was handled, otherwise UNHANDLED.

Raises:

Type	Description
ValueError	If data cannot be None.

Source code in src/pyromax/dispatcher/Router.py

async def notify(
    self,
    update: MaxObject,
    data: dict[Any, Any] | None = None,
    event_types: list[str] | None = None,
) -> Any:
    """Propagate an update through handlers and child routers.

    :param update: Incoming update object.
    :type update: MaxObject
    :param data: Context data available to handlers.
    :type data: dict[Any, Any] | None
    :param event_types: keys of Router.events
    :type event_types: list[str] | None

    :returns: Any if the update was handled, otherwise UNHANDLED.
    :rtype: Any

    :raises ValueError: If data cannot be None.
    """

    if event_types is None:
        event_types = []

    event_types = event_types[:]

    if data is None:
        raise ValueError("data cannot be None")

    unknown_update = False

    for key, event in self.events.items():
        if await event.is_my_update(update):
            if key not in event_types:
                event_types.append(key)
    if not event_types:
        unknown_update = True

    response = UNHANDLED

    for event_type in event_types:
        observer = self.events.get(event_type)
        if observer:
            result = await observer.check_root_filters(update, data)
            if not result:
                continue
            response = await observer.wrap_outer_middleware(
                observer.update,
                update,
                data=data,
            )
            if response is not UNHANDLED:
                return response

    if not self.sub_routers and unknown_update:
        return UNKNOWN_UPDATE

    if not self.sub_routers and not unknown_update:
        return UNHANDLED

    update_type_in_sub_routers = False
    for router in self.sub_routers:
        response = await router.notify(
            update=update, data=data, event_types=event_types
        )
        if response is UNHANDLED:
            update_type_in_sub_routers = True

        if response not in (UNKNOWN_UPDATE, UNHANDLED):
            return response
    else:
        if update_type_in_sub_routers:
            return UNHANDLED
        return UNKNOWN_UPDATE