Filters

Filter

Filter()

Bases: ABC

Base class for user-defined and built-in filters.

Subclasses must implement the asynchronous _check method and define which update types they accept through work_with.

Initialize the filter.

Source code in src/pyromax/filters/base.py

def __init__(self) -> None:
    """Initialize the filter.
    """
    self._logger = logging.getLogger(f"{self.__class__.__name__}")

work_with abstractmethod property

work_with: tuple[type[BaseMaxObject], ...]

Work with.

Returns:

Type	Description
tuple[type[BaseMaxObject], ...]	The resulting tuple[type[BaseMaxObject], ...] value.

callback property

callback: Callable[..., Awaitable[bool | dict[str, Any]]]

Callback.

Parameters:

Name	Type	Description	Default
args	Any	Positional arguments forwarded to the wrapped callable.	required
kwargs	Any	Keyword arguments forwarded to the wrapped callable.	required

Returns:

Type	Description
Callable[..., Awaitable[bool | dict[str, Any]]]	The resulting Callable[..., Awaitable[bool | dict[str, Any]]] value.

__call__ async

__call__(
    update: ResolvedUpdate,
    data: dict[Any, Any],
    *args: Any,
    **kwargs: Any
) -> bool | dict[str, Any]

Return whether the event matches the filter.

Parameters:

Name	Type	Description	Default
update	ResolvedUpdate	Incoming update to process.	required
data	dict[Any, Any]	Contextual data passed through the processing pipeline.	required
args	Any	Positional arguments forwarded to the wrapped callable.	()
kwargs	Any	Keyword arguments forwarded to the wrapped callable.	{}

Returns:

Type	Description
bool | dict[str, Any]	The resulting bool | dict[str, Any] value.

Source code in src/pyromax/filters/base.py

async def __call__(
    self, update: ResolvedUpdate, data: dict[Any, Any], *args: Any, **kwargs: Any
) -> bool | dict[str, Any]:
    """Return whether the event matches the  filter.

    :param update: Incoming update to process.
    :type update: ResolvedUpdate
    :param data: Contextual data passed through the processing pipeline.
    :type data: dict[Any, Any]
    :param args: Positional arguments forwarded to the wrapped callable.
    :type args: Any
    :param kwargs: Keyword arguments forwarded to the wrapped callable.
    :type kwargs: Any
    :returns: The resulting bool | dict[str, Any] value.
    :rtype: bool | dict[str, Any]
    """
    if self._SKIP_CHECK_PREPARATIONS:
        return await self._check(update, data, *args, **kwargs)

    if not type(update) in self.work_with:
        return False

    data.update({type(elem): elem for elem in args})

    data.update(kwargs)

    check_args = inspect_and_form(self.callback, data=data)

    return await self._check(**check_args)

__invert__

__invert__() -> Filter

Invert.

Returns:

Type	Description
Filter	The resulting Filter value.

Source code in src/pyromax/filters/base.py

def __invert__(self) -> Filter:
    """Invert.

    :returns: The resulting Filter value.
    :rtype: Filter
    """
    from .logic import invert_f

    return invert_f(self)

Command

Command(
    *values: Any,
    commands: list[str] | None = None,
    prefix: str = "/",
    ignore_case: bool = False,
    ignore_mention: bool = False,
    magic: None = None
)

Bases: Filter

This filter can be helpful for handling commands from the text messages.

Works only with :class:aiogram.types.message.Message events which have the :code:text.

List of commands (string or compiled regexp patterns)

Parameters:

Name	Type	Description	Default
prefix	str	Prefix for command. Prefix is always a single char but here you can pass all of allowed prefixes, for example: :code:"/!" will work with commands prefixed by :code:"/" or :code:"!".	'/'
ignore_case	bool	Ignore case (Does not work with regexp, use flags instead)	False
ignore_mention	bool	Ignore bot mention. By default, bot can not handle commands intended for other bots	False
magic	None	Validate command object via Magic filter after all checks done	None
values	Any	Values to validate or transform.	()
commands	list[str] | None	Collection of commands.	None

Raises:

Type	Description
ValueError	If the requested action cannot be completed.

Source code in src/pyromax/filters/Command.py

def __init__(
    self,
    *values: Any,
    commands: list[str] | None = None,
    prefix: str = "/",
    ignore_case: bool = False,
    ignore_mention: bool = False,
    magic: None = None,
) -> None:
    """List of commands (string or compiled regexp patterns)

    :param prefix: Prefix for command.
        Prefix is always a single char but here you can pass all of allowed prefixes,
        for example: :code:`"/!"` will work with commands prefixed
        by :code:`"/"` or :code:`"!"`.
    :param ignore_case: Ignore case (Does not work with regexp, use flags instead)
    :param ignore_mention: Ignore bot mention. By default,
        bot can not handle commands intended for other bots
    :param magic: Validate command object via Magic filter after all checks done

    :param values: Values to validate or transform.
    :type values: Any
    :param commands: Collection of commands.
    :type commands: list[str] | None
    :type prefix: str
    :type ignore_case: bool
    :type ignore_mention: bool
    :type magic: None
    :raises ValueError: If the requested action cannot be completed.
    """
    super().__init__()
    if commands is None:
        commands = []
    if isinstance(commands, (str, Pattern)):
        commands = [commands]

    if not isinstance(commands, Iterable):
        msg = "Command filter only supports str, re.Pattern, BotCommand object or their Iterable"
        raise ValueError(msg)

    items = []
    for command in (*values, *commands):
        # if isinstance(command, BotCommand):
        #     command = command.command
        if not isinstance(command, (str, Pattern)):
            msg = (
                "Command filter only supports str, re.Pattern, BotCommand object"
                " or their Iterable"
            )
            raise ValueError(msg)
        if ignore_case and isinstance(command, str):
            command = command.casefold()
        items.append(command)

    if not items:
        msg = "At least one command should be specified"
        raise ValueError(msg)

    self.commands = tuple(items)
    self.prefix = prefix
    self.ignore_case = ignore_case
    self.ignore_mention = ignore_mention
    self.magic = magic

work_with property

work_with: tuple[type[Message]]

Work with.

Returns:

Type	Description
tuple[type[Message]]	The resulting tuple[type[Message]] value.

__slots__ class-attribute instance-attribute

__slots__ = (
    "commands",
    "ignore_case",
    "ignore_mention",
    "magic",
    "prefix",
)

commands instance-attribute

commands = tuple(items)

prefix instance-attribute

prefix = prefix

ignore_case instance-attribute

ignore_case = ignore_case

ignore_mention instance-attribute

ignore_mention = ignore_mention

magic instance-attribute

magic = magic

extract_command classmethod

extract_command(text: str) -> CommandObject

Extract command.

Parameters:

Name	Type	Description	Default
text	str	Message or textual content.	required

Returns:

Type	Description
CommandObject	The resulting CommandObject value.

Raises:

Type	Description
CommandException	If the requested action cannot be completed.

Source code in src/pyromax/filters/Command.py

@classmethod
def extract_command(cls, text: str) -> CommandObject:
    # First step: separate command with arguments
    # "/command@mention arg1 arg2" -> "/command@mention", ["arg1 arg2"]
    """Extract command.

    :param text: Message or textual content.
    :type text: str
    :returns: The resulting CommandObject value.
    :rtype: CommandObject
    :raises CommandException: If the requested action cannot be completed.
    """
    try:
        full_command, *args = text.split(maxsplit=1)
    except ValueError as e:
        msg = "not enough values to unpack"
        raise CommandException(msg) from e

    # Separate command into valuable parts
    # "/command@mention" -> "/", ("command", "@", "mention")
    prefix, (command, _, mention) = full_command[0], full_command[1:].partition("@")
    return CommandObject(
        prefix=prefix,
        command=command,
        mention=mention or None,
        args=args[0] if args else None,
    )

validate_prefix

validate_prefix(command: CommandObject) -> None

Validate prefix.

Parameters:

Name	Type	Description	Default
command	CommandObject	CommandObject instance to process.	required

Raises:

Type	Description
CommandException	If the requested action cannot be completed.

Source code in src/pyromax/filters/Command.py

def validate_prefix(self, command: CommandObject) -> None:
    """Validate prefix.

    :param command: CommandObject instance to process.
    :type command: CommandObject
    :raises CommandException: If the requested action cannot be completed.
    """
    if command.prefix not in self.prefix:
        msg = "Invalid command prefix"
        raise CommandException(msg)

validate_command

validate_command(command: CommandObject) -> CommandObject

Validate command.

Parameters:

Name	Type	Description	Default
command	CommandObject	CommandObject instance to process.	required

Returns:

Type	Description
CommandObject	The resulting CommandObject value.

Raises:

Type	Description
CommandException	If the requested action cannot be completed.

Source code in src/pyromax/filters/Command.py

def validate_command(self, command: CommandObject) -> CommandObject:
    """Validate command.

    :param command: CommandObject instance to process.
    :type command: CommandObject
    :returns: The resulting CommandObject value.
    :rtype: CommandObject
    :raises CommandException: If the requested action cannot be completed.
    """
    for allowed_command in cast(Sequence[CommandPatternType], self.commands):
        # Command can be presented as regexp pattern or raw string
        # then need to validate that in different ways
        if isinstance(allowed_command, Pattern):  # Regexp
            result = allowed_command.match(command.command)
            if result:
                return replace(command, regexp_match=result)

        command_name = command.command
        if self.ignore_case:
            command_name = command_name.casefold()

        if command_name == allowed_command:  # String
            return command
    msg = "Command did not match pattern"
    raise CommandException(msg)

parse_command async

parse_command(text: str, max_api: MaxApi) -> CommandObject

Extract command from the text and validate

Parameters:

Name	Type	Description	Default
text	str		required
max_api	MaxApi		required

Returns:

Type	Description
CommandObject	The resulting CommandObject value.

Source code in src/pyromax/filters/Command.py

async def parse_command(self, text: str, max_api: MaxApi) -> CommandObject:
    """Extract command from the text and validate

    :param text:
    :param max_api:
    :return:

    :type text: str
    :type max_api: MaxApi
    :returns: The resulting CommandObject value.
    :rtype: CommandObject
    """
    command = self.extract_command(text)
    self.validate_prefix(command=command)
    # await self.validate_mention(bot=max_api, command=command)
    command = self.validate_command(command)
    # command = self.do_magic(command=command)
    return command

CommandObject dataclass

CommandObject(
    prefix: str = "/",
    command: str = "",
    mention: str | None = None,
    args: str | None = None,
    regexp_match: Match[str] | None = None,
    magic_result: Any | None = None,
)

Parsed command data extracted from a message.

The object stores command metadata such as prefix, command name, optional mention, and arguments.

prefix class-attribute instance-attribute

prefix: str = '/'

Command prefix

command class-attribute instance-attribute

command: str = ''

Command without prefix and mention

mention class-attribute instance-attribute

mention: str | None = None

Mention (if available)

args class-attribute instance-attribute

args: str | None = field(repr=False, default=None)

Command argument

regexp_match class-attribute instance-attribute

regexp_match: Match[str] | None = field(
    repr=False, default=None
)

Will be presented match result if the command is presented as regexp in filter

magic_result class-attribute instance-attribute

magic_result: Any | None = field(repr=False, default=None)

mentioned property

mentioned: bool

This command has mention?

Returns:

Type	Description
bool	True when the requested condition is satisfied; otherwise False.

text property

text: str

Generate original text from object

Returns:

Type	Description
str	The resulting str value.

MessageFilters

FromMeFilter

FromMeFilter()

Bases: Filter

Match updates that were sent by the current user.

Source code in src/pyromax/filters/base.py

def __init__(self) -> None:
    """Initialize the filter.
    """
    self._logger = logging.getLogger(f"{self.__class__.__name__}")

work_with property

work_with: tuple[type[Message]]

Work with.

Returns:

Type	Description
tuple[type[Message]]	The resulting tuple[type[Message]] value.

ReplyToMessageFilter

ReplyToMessageFilter()

Bases: Filter

Match messages that are replies to another message.

Source code in src/pyromax/filters/base.py

def __init__(self) -> None:
    """Initialize the filter.
    """
    self._logger = logging.getLogger(f"{self.__class__.__name__}")

work_with property

work_with: tuple[type[Message]]

Work with.

Returns:

Type	Description
tuple[type[Message]]	The resulting tuple[type[Message]] value.

MessageForwardFromFilter

MessageForwardFromFilter()

Bases: Filter

Match forwarded messages.

Source code in src/pyromax/filters/base.py

def __init__(self) -> None:
    """Initialize the filter.
    """
    self._logger = logging.getLogger(f"{self.__class__.__name__}")

work_with property

work_with: tuple[type[Message]]

Work with.

Returns:

Type	Description
tuple[type[Message]]	The resulting tuple[type[Message]] value.

MessageRemovedFilter

MessageRemovedFilter()

Bases: Filter

Match removed messages.

Source code in src/pyromax/filters/base.py

def __init__(self) -> None:
    """Initialize the filter.
    """
    self._logger = logging.getLogger(f"{self.__class__.__name__}")

work_with property

work_with: tuple[type[Message]]

Work with.

Returns:

Type	Description
tuple[type[Message]]	The resulting tuple[type[Message]] value.

FromChatFilter

FromChatFilter(chat_ids: int | Iterable[int])

Bases: Filter

Match messages from a specific chat.

Initialize the from chat filter.

Parameters:

Name	Type	Description	Default
chat_ids	int | Iterable[int]	Identifiers of the chats.	required

Source code in src/pyromax/filters/MessageFilters.py

def __init__(self, chat_ids: int | Iterable[int]) -> None:
    """Initialize the from chat filter.

    :param chat_ids: Identifiers of the chats.
    :type chat_ids: int | Iterable[int]
    """
    super().__init__()
    if isinstance(chat_ids, int):
        chat_ids = (chat_ids,)
    self.chat_ids = chat_ids

chat_ids instance-attribute

chat_ids = chat_ids

work_with property

work_with: tuple[type[Message]]

Work with.

Returns:

Type	Description
tuple[type[Message]]	The resulting tuple[type[Message]] value.

HaveAttachFilter

HaveAttachFilter()

Bases: Filter

Match messages that contain attachments.

Source code in src/pyromax/filters/base.py

def __init__(self) -> None:
    """Initialize the filter.
    """
    self._logger = logging.getLogger(f"{self.__class__.__name__}")

work_with property

work_with: tuple[type[Message]]

Work with.

Returns:

Type	Description
tuple[type[Message]]	The resulting tuple[type[Message]] value.

EmojiReactionFilters

EmojiReactionAddFilter

EmojiReactionAddFilter()

Bases: Filter

Match emoji reaction add events.

Source code in src/pyromax/filters/base.py

def __init__(self) -> None:
    """Initialize the filter.
    """
    self._logger = logging.getLogger(f"{self.__class__.__name__}")

work_with property

work_with: tuple[type[EmojiReaction]]

Work with.

Returns:

Type	Description
tuple[type[EmojiReaction]]	The resulting tuple[type[EmojiReaction]] value.

EmojiReactionRemoveFilter

EmojiReactionRemoveFilter()

Bases: Filter

Match emoji reaction remove events.

Source code in src/pyromax/filters/base.py

def __init__(self) -> None:
    """Initialize the filter.
    """
    self._logger = logging.getLogger(f"{self.__class__.__name__}")

work_with property

work_with: tuple[type[EmojiReaction]]

Work with.

Returns:

Type	Description
tuple[type[EmojiReaction]]	The resulting tuple[type[EmojiReaction]] value.