Improved dispatching for Pyrogram
from . import filters
from .dispatcher import Dispatcher, RunLogic
from .expected_conditions import ExpectedCondition
from .router import Router
from .filters import Filter
from . import utils, union_types, handlers

__version__ = "0.1.0"

dispyro/__init__.py
asteroid-den/dispyro
from typing import Any, Dict, List

from pyrogram import Client, handlers, idle
from pyrogram.handlers.handler import Handler

from .enums import RunLogic
from .router import Router
from .union_types import Update


class Dispatcher:
    """Main class to interract with API. Dispatcher itself cannot be used to
    register handlers, so routers must be attached to it.
    """

    def __init__(
        self,
        *clients: Client,
        ignore_preparation: bool = False,
        clear_on_prepare: bool = True,
        run_logic: RunLogic = RunLogic.ONE_RUN_PER_EVENT,
        **deps
    ):
        self.routers: List[Router] = []
        self._clients: List[Client] = []
        self._deps: Dict[str, Any] = deps

        self._ignore_preparation = ignore_preparation
        self._clear_on_prepare = clear_on_prepare
        self._run_logic = run_logic

        if ignore_preparation:
            self._clients = list(clients)

        else:
            for client in clients:
                client = self.prepare_client(
                    client=client, clear_handlers=clear_on_prepare
                )
                self._clients.append(client)

    def prepare_client(self, client: Client, clear_handlers: bool = True) -> Client:
        async def handler(client: Client, update: Update):
            await self.feed_update(client=client, update=update)

        handler_types: List[Handler] = [
            handlers.CallbackQueryHandler,
            handlers.ChatMemberUpdatedHandler,
            handlers.ChosenInlineResultHandler,
            handlers.EditedMessageHandler,
            handlers.InlineQueryHandler,
            handlers.MessageHandler,
            handlers.PollHandler,
        ]

        group = 0

        if clear_handlers:
            client.dispatcher.groups.clear()

        else:
            groups = list(client.dispatcher.groups.keys())

            if groups:
                group = max(groups) + 1

        for handler_type in handler_types:
            client.add_handler(handler_type(handler), group=group)

        return client

    def add_router(self, router: Router):
        self.routers.append(router)

    def add_routers(self, *routers: Router):
        self.routers.extend(routers)

    def cleanup(self) -> None:
        for router in self.routers:
            router.cleanup()

    async def feed_update(self, client: Client, update: Update) -> None:
        for router in self.routers:
            result = await router.feed_update(
                client=client, dispatcher=self, update=update, **self._deps
            )

            if self._run_logic is RunLogic.ONE_RUN_PER_EVENT and result:
                break

        self.cleanup()

    async def start(
        self,
        *clients: Client,
        ignore_preparation: bool = None,
        only_start: bool = False
    ) -> None:
        self.cleanup()

        if ignore_preparation is None:
            ignore_preparation = self._ignore_preparation

        if ignore_preparation:
            clients = list(clients)

        else:
            clients = [
                self.prepare_client(
                    client=client, clear_handlers=self._clear_on_prepare
                )
                for client in clients
            ]

        clients = self._clients + clients

        for client in clients:
            if not client.is_connected:
                await client.start()

        if not only_start:
            await idle()

dispyro/dispatcher.py
asteroid-den/dispyro
from enum import Enum, auto


class RunLogic(Enum):
    ONE_RUN_PER_EVENT = auto()
    ONE_RUN_PER_ROUTER = auto()
    UNLIMITED = auto()


class ExpectedConditionCheckType(Enum):
    ITER_ALL_ROUTERS = auto()
    PUT_ACTUAL_ROUTER = auto()

dispyro/enums.py
asteroid-den/dispyro
# This file defines core of filtering utility named `expected conditions`
# (abbreviated EC). This utility provides you ability to specify custom
# dispatch logic based on dispatching system state. You can find more detailed
# info in classes and functions docstrings.

from typing import Optional, Union

from . import utils
from .dispatcher import Dispatcher
from .router import Router
from .signatures import FilterCallback
from .union_types import Update
from .enums import ExpectedConditionCheckType


# too long notation when mentioning enum itself
ITER_ALL_ROUTERS = ExpectedConditionCheckType.ITER_ALL_ROUTERS
PUT_ACTUAL_ROUTER = ExpectedConditionCheckType.PUT_ACTUAL_ROUTER

RouterLike = Union[Router, str]


class ExpectedCondition:
    """Defines specific condition checked before router triggering."""

    def __init__(
        self,
        condition_callback: FilterCallback,
        check_type: ExpectedConditionCheckType,
        router: Router = None,
    ):
        if check_type is PUT_ACTUAL_ROUTER and router is None:
            raise ValueError("you must provide actual router for this check type")

        self._callback: FilterCallback = utils.safe_call(callable=condition_callback)
        self._check_type = check_type
        self._router = router

    def __invert__(self) -> "InvertedExpectedCondition":
        return InvertedExpectedCondition(
            condition_callback=self._callback,
            check_type=self._check_type,
            router=self._router,
        )

    async def check(self, dispatcher: Dispatcher, update: Update) -> bool:
        client = dispatcher.client
        deps = dispatcher._deps
        if self._check_type is ITER_ALL_ROUTERS:
            for router in dispatcher.routers:
                result = await self._callback(client, update, router, **deps)

                if result:
                    return result

            return False

        elif self._check_type is PUT_ACTUAL_ROUTER:
            return await self._callback(client, update, self._router, **deps)


class InvertedExpectedCondition(ExpectedCondition):
    def __invert__(self) -> ExpectedCondition:
        return ExpectedCondition(
            condition_callback=self._callback,
            check_type=self._check_type,
            router=self._router,
        )

    async def check(self, dispatcher: Dispatcher, update: Update) -> bool:
        client = dispatcher.client
        deps = dispatcher._deps
        if self._check_type == ITER_ALL_ROUTERS:
            for router in dispatcher.routers:
                result = not await self._callback(client, update, router)

                if result:
                    return result

            return True

        elif self._check_type == PUT_ACTUAL_ROUTER:
            return not await self._callback(client, update, self._router, **deps)


def router_is_triggered(related_router: RouterLike) -> ExpectedCondition:
    """Makes `ExpectedCondition` object which throws `True` if specified router
    was triggered during processing current update.

    IMPORTANT: if specified router goes after router that examines this condition,
    it won't work. Trigger check can only be performed backwards according to
    checks sequence.
    """
    router_id: Optional[int] = None
    router_name: Optional[str] = None

    if type(related_router) is Router:
        router_id = id(related_router)

    elif type(related_router) is str:
        router_name = related_router

        if router_name == "unnamed_router":
            raise ValueError("only named routers can be mentioned")

    else:
        raise ValueError(
            f"`related_router` type can be only `str` or `Router`, got `{type(related_router)}`"
        )

    async def callback(_, __, router: Router) -> bool:
        nonlocal router_id, router_name

        if router_id:
            return id(router) == router_id and router._triggered

        elif router_name:
            return router._name == router_name and router._triggered

    return ExpectedCondition(
        condition_callback=callback, check_type=ITER_ALL_ROUTERS, router=related_router
    )


def router_is_not_triggered(related_router: RouterLike) -> InvertedExpectedCondition:
    return ~router_is_triggered(related_router=related_router)

dispyro/expected_conditions.py
asteroid-den/dispyro
from pyrogram import Client
from pyrogram.filters import Filter as PyrogramFilter

from .signatures import FilterCallback
from .union_types import AnyFilter, Update
from .utils import safe_call


class Filter:
    """Custom version of `Filter` type, which supports DI. This filters can be
    combined with default pyrogram filters.
    """

    async def _default_callback(self, client: Client, update: Update):
        return True

    def __init__(self, callback: FilterCallback = None):
        self._unwrapped_callback = callback
        self._callback: FilterCallback = safe_call(callback or self._default_callback)

    async def __call__(self, client: Client, update: Update, **deps) -> bool:
        return await self._callback(client, update, **deps)

    def __invert__(self) -> "InvertedFilter":
        return InvertedFilter(callback=self._unwrapped_callback)

    def __and__(self, other: AnyFilter) -> "AndFilter":
        return AndFilter(left=self, right=other)

    def __or__(self, other: AnyFilter) -> "OrFilter":
        return OrFilter(left=self, right=other)


class InvertedFilter(Filter):
    async def __call__(self, client: Client, update: Update, **deps) -> bool:
        return not await super().__call__(client, update, **deps)


class AndFilter(Filter):
    def __init__(self, left: AnyFilter, right: AnyFilter):
        if isinstance(left, PyrogramFilter):
            left = safe_call(callable=left)

        if isinstance(right, PyrogramFilter):
            right = safe_call(callable=right)

        self._left: AnyFilter = left
        self._right: AnyFilter = right

    async def __call__(self, client: Client, update: Update, **deps) -> bool:
        left_value = await self._left(client, update, **deps)

        if not left_value:
            return False

        right_value = await self._right(client, update, **deps)

        return left_value and right_value


class OrFilter(Filter):
    def __init__(self, left: AnyFilter, right: AnyFilter):
        if isinstance(left, PyrogramFilter):
            left = safe_call(callable=left)

        if isinstance(right, PyrogramFilter):
            right = safe_call(callable=right)

        self._left: AnyFilter = left
        self._right: AnyFilter = right

    async def __call__(self, client: Client, update: Update, **deps) -> bool:
        left_value = await self._left(client, update, **deps)

        if left_value:
            return True

        right_value = await self._right(client, update, **deps)

        return left_value or right_value

dispyro/filters.py
asteroid-den/dispyro
# This file defines custom handlers definitions that is used instead of
# original Pyrogram handlers. Unlike original handlers, customs are DI-friendly,
# providing ability to use dependency injection with ease.
#
# Groups were removed because of changed processing logic (actual handlers
# distribution instead of formal split by groups, which made groups useless).
# Instead of them, new attribute added: `priority`, which affects on handlers
# order inside of specific router. Lower number means higher priority (highest
# priority is 1), handlers with same priority arranged corresponding to order
# in which they were registered. By default, all registered handlers priority
# is set to 1. You can customize this behaviour by providing your own
# `priority_factory`. This function must take 2 arguments
# (handler itself and `router` that registering this handler) and return
# positive `int`.

from typing import Callable

from pyrogram import Client, types

import dispyro

from .signatures import (
    CallbackQueryHandlerCallback,
    ChatMemberUpdatedHandlerCallback,
    ChosenInlineResultHandlerCallback,
    EditedMessageHandlerCallback,
    InlineQueryHandlerCallback,
    MessageHandlerCallback,
    PollHandlerCallback,
)
from .union_types import AnyFilter, Callback, Update
from .utils import safe_call

PriorityFactory = Callable[["Handler", "dispyro.Router"], int]


class Handler:
    def _default_priority_factory(self, _) -> int:
        return 1

    _priority_factory: PriorityFactory = _default_priority_factory

    @classmethod
    def set_priority_factory(cls, priority_factory: PriorityFactory) -> None:
        cls._priority_factory = priority_factory

    def __init__(
        self,
        *,
        callback: Callback,
        router: "dispyro.Router",
        name: str = None,
        priority: int = None,
        filters: AnyFilter = dispyro.filters.Filter(),
    ):
        if priority is not None:
            self._priority = priority
        else:
            self._priority = self._priority_factory(router)

        self._name = name or "unnamed_handler"
        self.callback: Callback = safe_call(callable=callback)
        self._router = router
        self._filters: dispyro.filters.Filter = dispyro.filters.Filter() & filters

        # This field indicates whether handler was called during handling current
        # update. Defaults to `False`. Set to `False` on cleanup (after finishing
        # update processing).
        self._triggered: bool = False

    async def __call__(
        self,
        *,
        client: Client,
        update: Update,
        **deps,
    ) -> None:
        filters_passed = await self._filters(client=client, update=update, **deps)

        if not filters_passed:
            return

        await self.callback(client, update, **deps)
        self._triggered = True

    def __repr__(self) -> str:
        return f"{self.__class__.__name__} `{self._name}`"


class CallbackQueryHandler(Handler):
    callback: CallbackQueryHandlerCallback

    def __init__(
        self,
        *,
        callback: CallbackQueryHandlerCallback,
        router: "dispyro.Router",
        name: str = None,
        priority: int = None,
        filters: AnyFilter = dispyro.filters.Filter(),
    ):
        super().__init__(
            callback=callback,
            router=router,
            name=name,
            priority=priority,
            filters=filters,
        )

    async def __call__(
        self,
        *,
        client: Client,
        update: types.CallbackQuery,
        **deps,
    ) -> None:
        await super().__call__(client=client, update=update, **deps)


class ChatMemberUpdatedHandler(Handler):
    callback: ChatMemberUpdatedHandlerCallback

    def __init__(
        self,
        *,
        callback: ChatMemberUpdatedHandlerCallback,
        router: "dispyro.Router",
        name: str = None,
        priority: int = None,
        filters: AnyFilter = dispyro.filters.Filter(),
    ):
        super().__init__(
            callback=callback,
            router=router,
            name=name,
            priority=priority,
            filters=filters,
        )

    async def __call__(
        self,
        *,
        client: Client,
        update: types.ChatMemberUpdated,
        **deps,
    ) -> None:
        await super().__call__(client=client, update=update, **deps)


class ChosenInlineResultHandler(Handler):
    callback: ChosenInlineResultHandlerCallback

    def __init__(
        self,
        *,
        callback: ChosenInlineResultHandlerCallback,
        router: "dispyro.Router",
        name: str = None,
        priority: int = None,
        filters: AnyFilter = dispyro.filters.Filter(),
    ):
        super().__init__(
            callback=callback,
            router=router,
            name=name,
            priority=priority,
            filters=filters,
        )

    async def __call__(
        self,
        *,
        client: Client,
        update: types.ChosenInlineResult,
        **deps,
    ) -> None:
        await super().__call__(client=client, update=update, **deps)


class EditedMessageHandler(Handler):
    callback: EditedMessageHandlerCallback

    def __init__(
        self,
        *,
        callback: EditedMessageHandlerCallback,
        router: "dispyro.Router",
        name: str = None,
        priority: int = None,
        filters: AnyFilter = dispyro.filters.Filter(),
    ):
        super().__init__(
            callback=callback,
            router=router,
            name=name,
            priority=priority,
            filters=filters,
        )

    async def __call__(
        self,
        *,
        client: Client,
        update: types.Message,
        **deps,
    ) -> None:
        await super().__call__(client=client, update=update, **deps)


class InlineQueryHandler(Handler):
    callback: InlineQueryHandlerCallback

    def __init__(
        self,
        *,
        callback: InlineQueryHandlerCallback,
        router: "dispyro.Router",
        name: str = None,
        priority: int = None,
        filters: AnyFilter = dispyro.filters.Filter(),
    ):
        super().__init__(
            callback=callback,
            router=router,
            name=name,
            priority=priority,
            filters=filters,
        )

    async def __call__(
        self,
        *,
        client: Client,
        update: types.InlineQuery,
        **deps,
    ) -> None:
        await super().__call__(client=client, update=update, **deps)


class MessageHandler(Handler):
    callback: MessageHandlerCallback

    def __init__(
        self,
        *,
        callback: MessageHandlerCallback,
        router: "dispyro.Router",
        name: str = None,
        priority: int = None,
        filters: AnyFilter = dispyro.filters.Filter(),
    ):
        super().__init__(
            callback=callback,
            router=router,
            name=name,
            priority=priority,
            filters=filters,
        )

    async def __call__(
        self,
        *,
        client: Client,
        update: types.Message,
        **deps,
    ) -> None:
        await super().__call__(client=client, update=update, **deps)


class PollHandler(Handler):
    callback: PollHandlerCallback

    def __init__(
        self,
        *,
        callback: PollHandlerCallback,
        router: "dispyro.Router",
        name: str = None,
        priority: int = None,
        filters: AnyFilter = dispyro.filters.Filter(),
    ):
        super().__init__(
            callback=callback,
            router=router,
            name=name,
            priority=priority,
            filters=filters,
        )

    async def __call__(
        self,
        *,
        client: Client,
        update: types.Poll,
        **deps,
    ) -> None:
        await super().__call__(client=client, update=update, **deps)

dispyro/handlers.py
asteroid-den/dispyro
from typing import Callable, List

from pyrogram import Client, types

import dispyro

from .enums import RunLogic
from .filters import Filter
from .handlers import (
    CallbackQueryHandler,
    ChatMemberUpdatedHandler,
    ChosenInlineResultHandler,
    EditedMessageHandler,
    InlineQueryHandler,
    MessageHandler,
    PollHandler,
)
from .union_types import AnyFilter, Callback, Handler, Update


class HandlersHolder:
    __handler_type__: Handler

    def __init__(self, router: "dispyro.Router", filters: AnyFilter = None):
        self.filters = Filter() & filters if filters else Filter()
        self.handlers: List[Handler] = []
        self._router = router

    def filter(self, filter: AnyFilter) -> None:
        self.filters &= filter

    def __call__(
        self, filters: Filter = Filter(), priority: int = None
    ) -> Callable[[Callback], Callback]:
        def decorator(callback: Callback) -> Callback:
            nonlocal filters

            handler_type = self.__handler_type__
            self.handlers.append(
                handler_type(
                    callback=callback,
                    router=self._router,
                    priority=priority,
                    filters=filters,
                )
            )

            return callback

        return decorator

    async def feed_update(
        self, client: Client, run_logic: RunLogic, update: Update, **deps
    ) -> bool:
        filters_passed = await self.filters(client=client, update=update, **deps)

        if not filters_passed:
            return

        self._router._triggered = True

        handlers = sorted(self.handlers, key=lambda x: x._priority)
        for handler in handlers:
            await handler(client=client, update=update, **deps)

            if handler._triggered and run_logic in {
                RunLogic.ONE_RUN_PER_ROUTER,
                RunLogic.ONE_RUN_PER_EVENT,
            }:
                return True

        return any(handler._triggered for handler in handlers)

class CallbackQueryHandlersHolder(HandlersHolder):
    __handler_type__ = CallbackQueryHandler
    handlers: List[CallbackQueryHandler]

    async def feed_update(
        self, client: Client, run_logic: RunLogic, update: types.CallbackQuery, **deps
    ) -> bool:
        return await super().feed_update(
            client=client, run_logic=run_logic, update=update, **deps
        )


class ChatMemberUpdatedHandlersHolder(HandlersHolder):
    __handler_type__ = ChatMemberUpdatedHandler
    handlers: List[ChatMemberUpdatedHandler]

    async def feed_update(
        self,
        client: Client,
        run_logic: RunLogic,
        update: types.ChatMemberUpdated,
        **deps
    ) -> bool:
        return await super().feed_update(
            client=client, run_logic=run_logic, update=update, **deps
        )


class ChosenInlineResultHandlersHolder(HandlersHolder):
    __handler_type__ = ChosenInlineResultHandler
    handlers: List[ChosenInlineResultHandler]

    async def feed_update(
        self,
        client: Client,
        run_logic: RunLogic,
        update: types.ChosenInlineResult,
        **deps
    ) -> bool:
        return await super().feed_update(
            client=client, run_logic=run_logic, update=update, **deps
        )


class EditedMessageHandlersHolder(HandlersHolder):
    __handler_type__ = EditedMessageHandler
    handlers: List[EditedMessageHandler]

    async def feed_update(
        self, client: Client, run_logic: RunLogic, update: types.Message, **deps
    ) -> bool:
        return await super().feed_update(
            client=client, run_logic=run_logic, update=update, **deps
        )


class InlineQueryHandlersHolder(HandlersHolder):
    __handler_type__ = InlineQueryHandler
    handlers: List[InlineQueryHandler]

    async def feed_update(
        self, client: Client, run_logic: RunLogic, update: types.InlineQuery, **deps
    ) -> bool:
        return await super().feed_update(
            client=client, run_logic=run_logic, update=update, **deps
        )


class MessageHandlersHolder(HandlersHolder):
    __handler_type__ = MessageHandler
    handlers: List[MessageHandler]

    async def feed_update(
        self, client: Client, run_logic: RunLogic, update: types.Message, **deps
    ) -> bool:
        return await super().feed_update(
            client=client, run_logic=run_logic, update=update, **deps
        )


class PollHandlersHolder(HandlersHolder):
    __handler_type__ = PollHandler
    handlers: List[PollHandler]

    async def feed_update(
        self, client: Client, run_logic: RunLogic, update: types.Poll, **deps
    ) -> bool:
        return await super().feed_update(
            client=client, run_logic=run_logic, update=update, **deps
        )

dispyro/handlers_holders.py
asteroid-den/dispyro
from typing import List, TypeVar, cast

from pyrogram import Client, types

import dispyro

from .handlers import Handler
from .handlers_holders import (
    CallbackQueryHandlersHolder,
    ChatMemberUpdatedHandlersHolder,
    ChosenInlineResultHandlersHolder,
    EditedMessageHandlersHolder,
    InlineQueryHandlersHolder,
    MessageHandlersHolder,
    PollHandlersHolder,
)
from .union_types import Update

ExpectedCondition = TypeVar("ExpectedCondition", bound="dispyro.ExpectedCondition")


class Router:
    """Router class used to put subset of handlers together.

    To put things to work, must be attached to `Dispatcher`.
    """

    def __init__(self, name: str = None):
        self._name = name or "unnamed_router"

        self.callback_query = CallbackQueryHandlersHolder(router=self)
        self.chat_member_updated = ChatMemberUpdatedHandlersHolder(router=self)
        self.chosen_inline_result = ChosenInlineResultHandlersHolder(router=self)
        self.edited_message = EditedMessageHandlersHolder(router=self)
        self.inline_query = InlineQueryHandlersHolder(router=self)
        self.message = MessageHandlersHolder(router=self)
        self.poll = PollHandlersHolder(router=self)

        self._expected_conditions: List[ExpectedCondition] = []

        # This field indicates whether any of router related handlers was
        # called during handling current update. Defaults to `False`. Set to
        # `False` on cleanup (after finishing update processing).
        self._triggered: bool = False

    def __repr__(self) -> str:
        return f"{self.__class__.__name__} `{self._name}`"

    @property
    def all_handlers(self) -> List[Handler]:
        return [
            *self.callback_query.handlers,
            *self.chat_member_updated.handlers,
            *self.chosen_inline_result.handlers,
            *self.edited_message.handlers,
            *self.inline_query.handlers,
            *self.message.handlers,
            *self.poll.handlers,
        ]

    def cleanup(self) -> None:
        self._triggered = False

        for handler in self.all_handlers:
            handler._triggered = False

    def add_condition(self, expected_condition: ExpectedCondition):
        self._expected_conditions.append(expected_condition)

    async def feed_update(
        self, client: Client, dispatcher: "dispyro.Dispatcher", update: Update, **deps
    ) -> bool:
        update_type = type(update)
        result = False
        run_logic = dispatcher._run_logic

        for expected_condition in self._expected_conditions:
            if not await expected_condition.check(dispatcher=dispatcher, update=update):
                return result

        if update_type is types.CallbackQuery:
            result = await self.callback_query.feed_update(
                client=client, run_logic=run_logic, update=update, **deps
            )

        elif update_type is types.ChatMemberUpdated:
            result = await self.chat_member_updated.feed_update(
                client=client, run_logic=run_logic, update=update, **deps
            )

        elif update_type is types.ChosenInlineResult:
            result = await self.chosen_inline_result.feed_update(
                client=client, run_logic=run_logic, update=update, **deps
            )

        elif update_type is types.Message:
            update = cast(types.Message, update)

            if update.edit_date:
                result = await self.edited_message.feed_update(
                    client=client, run_logic=run_logic, update=update, **deps
                )

            else:
                result = await self.message.feed_update(
                    client=client, run_logic=run_logic, update=update, **deps
                )

        elif update_type is types.InlineQuery:
            result = await self.inline_query.feed_update(
                client=client, run_logic=run_logic, update=update, **deps
            )

        elif update_type is types.Poll:
            result = await self.poll.feed_update(
                client=client, run_logic=run_logic, update=update, **deps
            )

        return result

dispyro/router.py
asteroid-den/dispyro
from .filter_callback import FilterCallback
from .handler_callback import (
    CallbackQueryHandlerCallback,
    ChatMemberUpdatedHandlerCallback,
    ChosenInlineResultHandlerCallback,
    EditedMessageHandlerCallback,
    InlineQueryHandlerCallback,
    MessageHandlerCallback,
    PollHandlerCallback,
)

__all__ = (
    "FilterCallback",
    "CallbackQueryHandlerCallback",
    "ChatMemberUpdatedHandlerCallback",
    "ChosenInlineResultHandlerCallback",
    "EditedMessageHandlerCallback",
    "InlineQueryHandlerCallback",
    "MessageHandlerCallback",
    "PollHandlerCallback",
)

dispyro/signatures/__init__.py
asteroid-den/dispyro
from typing import Protocol

from pyrogram import Client

import dispyro


class FilterCallback(Protocol):
    """Signature class for filters callback with DI support."""

    async def __call__(
        self, client: Client, update: "dispyro.union_types.Update", **deps
    ) -> bool:
        ...

dispyro/signatures/filter_callback.py
asteroid-den/dispyro
from typing import Any, Protocol

from pyrogram import Client
from pyrogram.types import (
    CallbackQuery,
    ChatMemberUpdated,
    ChosenInlineResult,
    InlineQuery,
    Message,
    Poll,
)


class CallbackQueryHandlerCallback(Protocol):
    """Signature class for CallbackQuery handler callback with DI support."""

    async def __call__(
        self, client: Client, callback_query: CallbackQuery, **deps
    ) -> Any:
        ...


class ChatMemberUpdatedHandlerCallback(Protocol):
    """Signature class for ChatMemberUpdated handler callback with DI support."""

    async def __call__(
        self, client: Client, chat_member_updated: ChatMemberUpdated, **deps
    ) -> Any:
        ...


class ChosenInlineResultHandlerCallback(Protocol):
    """Signature class for ChosenInlineResult handler callback with DI support."""

    async def __call__(
        self, client: Client, chosen_inline_result: ChosenInlineResult, **deps
    ) -> Any:
        ...


class EditedMessageHandlerCallback(Protocol):
    """Signature class for Message handler (handling edited messages) callback with DI support."""

    async def __call__(self, client: Client, message: Message, **deps) -> Any:
        ...


class InlineQueryHandlerCallback(Protocol):
    """Signature class for InlineQuery handler callback with DI support."""

    async def __call__(
        self, client: Client, inline_query: InlineQuery, **deps
    ) -> Any:
        ...


class MessageHandlerCallback(Protocol):
    """Signature class for Message handler callback with DI support."""

    async def __call__(self, client: Client, message: Message, **deps) -> Any:
        ...


class PollHandlerCallback(Protocol):
    """Signature class for Poll handler callback with DI support."""

    async def __call__(self, client: Client, poll: Poll, **deps) -> Any:
        ...

dispyro/signatures/handler_callback.py
asteroid-den/dispyro
from typing import Union  # I prefer X | Y notation, but for compability I'll use this.

import pyrogram
from pyrogram.types import (
    CallbackQuery,
    ChatMemberUpdated,
    ChosenInlineResult,
    InlineQuery,
    Message,
    Poll,
)

import dispyro

from .signatures.handler_callback import (
    CallbackQueryHandlerCallback,
    ChatMemberUpdatedHandlerCallback,
    ChosenInlineResultHandlerCallback,
    EditedMessageHandlerCallback,
    InlineQueryHandlerCallback,
    MessageHandlerCallback,
    PollHandlerCallback,
)
# from .handlers import (
#     CallbackQueryHandler,
#     ChatMemberUpdatedHandler,
#     ChosenInlineResultHandler,
#     EditedMessageHandler,
#     InlineQueryHandler,
#     MessageHandler,
#     PollHandler,
# )

Handler = Union[
    "dispyro.handlers.CallbackQueryHandler",
    "dispyro.handlers.ChatMemberUpdatedHandler",
    "dispyro.handlers.ChosenInlineResultHandler",
    "dispyro.handlers.EditedMessageHandler",
    "dispyro.handlers.InlineQueryHandler",
    "dispyro.handlers.MessageHandler",
    "dispyro.handlers.PollHandler",
]

Callback = Union[
    CallbackQueryHandlerCallback,
    ChatMemberUpdatedHandlerCallback,
    ChosenInlineResultHandlerCallback,
    EditedMessageHandlerCallback,
    InlineQueryHandlerCallback,
    MessageHandlerCallback,
    PollHandlerCallback,
]

Update = Union[
    CallbackQuery, ChatMemberUpdated, ChosenInlineResult, Message, InlineQuery, Poll
]

AnyFilter = Union[pyrogram.filters.Filter, "dispyro.filters.Filter"]

__all__ = ("Handler", "Callback", "Update", "AnyFilter")

dispyro/union_types.py
asteroid-den/dispyro
from functools import wraps
from typing import Any, Callable, Dict, TypeVar

ReturnType = TypeVar("ReturnType")


def get_needed_kwargs(callable: Callable, **kwargs) -> Dict[str, Any]:
    """Helper function fetches needed `kwargs`.
    Returns only needed kwargs in a form of a `dict`.
    """

    if not getattr(callable, "__code__", None):
        # treating `callable` as non-function object with __call__ defined
        callable = callable.__call__

    # using internals to check needed arguments
    varnames: tuple[str, ...] = callable.__code__.co_varnames
    needed_kwargs = {k: v for k, v in kwargs.items() if k in varnames}

    return needed_kwargs


def safe_call(callable: Callable[..., ReturnType]) -> Callable[..., ReturnType]:
    """Helper function that makes new `callable` which feeds only needed `kwargs` to original."""

    @wraps(callable)
    def wrapper(*args, **kwargs) -> ReturnType:
        needed_kwargs = get_needed_kwargs(callable=callable, **kwargs)
        return callable(*args, **needed_kwargs)

    return wrapper

dispyro/utils.py
asteroid-den/dispyro
import asyncio
from typing import List

from pyrogram import Client, filters, types

from dispyro import Dispatcher, Router

router = Router()
router.message.filter(filters.me)  # processing only messages from account itself


@router.message(filters.command("numbers", prefixes="."))
async def handler(_, message: types.Message, numbers: List[int]):
    formatted_numbers = ", ".join(map(str, numbers))
    text = f"Entered numbers: {formatted_numbers}"
    await message.edit_text(text=text)


@router.message(filters.command("add", prefixes="."))
async def handler(_, message: types.Message, numbers: List[int]):
    values = message.command[1:]

    if not values:
        await message.delete()
        return

    for value in values:
        if value.isdigit():
            value = int(value)
            numbers.append(value)

    await message.edit_text(text="✅ Numbers added")


@router.message(filters.command("clear", prefixes="."))
async def handler(_, message: types.Message, numbers: List[int]):
    numbers.clear()

    await message.edit_text(text="✅ Numbers list cleared")


async def main():
    client = Client(
        name="dispyro",
        api_id=2040,  # TDesktop api_id, better to be replaced with your value
        api_hash="b18441a1ff607e10a989891a5462e627",  # TDesktop api_hash, better to be replaced with your value
    )
    dispatcher = Dispatcher(client, numbers=[])
    dispatcher.add_router(router)

    await dispatcher.start()


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

examples/di.py
asteroid-den/dispyro
import asyncio
from typing import List

from pyrogram import Client, filters, types

from dispyro import Dispatcher, Router, Filter

router = Router()
router.message.filter(filters.me)  # processing only messages from account itself
router.message.filter(filters.command("salad", prefixes=".")) # processing only .salad command

async def fruits_filter_callback(_, message: types.Message, fruits: List[str]):
    # this filter goes after filters.command(), so message.command is not None
    return all(fruit in fruits for fruit in message.command[1:])

fruits_filter = Filter(fruits_filter_callback)

@router.message(fruits_filter)
async def handler(_, message: types.Message):
    await message.edit_text(text="😋 Yummy fruit salad")

@router.message(~fruits_filter)
async def handler(_, message: types.Message):
    await message.edit_text(text="😔 We don't have needed fruit")


async def main():
    client = Client(
        name="dispyro",
        api_id=2040,  # TDesktop api_id, better to be replaced with your value
        api_hash="b18441a1ff607e10a989891a5462e627",  # TDesktop api_hash, better to be replaced with your value
    )
    dispatcher = Dispatcher(client, fruits=["apple", "banana", "strawberry"])
    dispatcher.add_router(router)

    await dispatcher.start()


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

examples/filter_with_di.py
asteroid-den/dispyro
import asyncio

from pyrogram import Client, filters, types

from dispyro import Dispatcher, Router

router = Router()
router.message.filter(filters.me)  # processing only messages from account itself


@router.message(filters.command("ping", prefixes="."))
async def handler(_, message: types.Message):
    await message.edit_text(text="🏓 pong!")


async def main():
    client = Client(
        name="dispyro",
        api_id=2040,  # TDesktop api_id, better to be replaced with your value
        api_hash="b18441a1ff607e10a989891a5462e627",  # TDesktop api_hash, better to be replaced with your value
    )
    dispatcher = Dispatcher(client)
    dispatcher.add_router(router)

    await dispatcher.start()


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

examples/hello_world.py
asteroid-den/dispyro
import re

import setuptools

with open("README.md") as f:
    long_description = f.read()

with open("requirements.txt") as f:
    requires = f.read().split()

with open("dispyro/__init__.py") as f:
    version = re.search(r'__version__ = "(?P<version>\d+\.\d+\.\d+)"', f.read()).group(
        "version"
    )

setuptools.setup(
    name="dispyro",
    version=version,
    author="asteroid-den",
    author_email="denbartolomee@gmail.com",
    description="Improved dispatching for Pyrogram",
    long_description=long_description,
    long_description_content_type="text/markdown",
    url="https://github.com/asteroid-den/dispyro/",
    packages=setuptools.find_packages(),
    license="MIT",
    classifiers=[
        "Programming Language :: Python :: 3",
        "License :: OSI Approved :: MIT License",
        "Operating System :: OS Independent",
    ],
    install_requires=requires,
    python_requires=">=3.7",
)

setup.py
asteroid-den/dispyro