Package: maxapi.filters¶

`maxapi.filters` ¶

`BaseFilter` ¶

Базовый класс для фильтров.

Определяет интерфейс фильтрации событий. Потомки должны переопределять метод call.

Methods:

Name	Description
`__call__`	Асинхронная проверка события на соответствие фильтру.

`filter_attrs(obj, *filters)` ¶

Применяет один или несколько фильтров MagicFilter к объекту.

Parameters:

Name	Type	Description	Default
`obj`	`object`	Объект, к которому применяются фильтры (например, event или message).	required
`*filters`	`MagicFilter`	Один или несколько выражений MagicFilter.	`()`

Returns:

Name	Type	Description
`bool`	`bool`	True, если все фильтры возвращают True, иначе False.

Source code in maxapi/filters/__init__.py

def filter_attrs(obj: object, *filters: MagicFilter) -> bool:
    """
    Применяет один или несколько фильтров MagicFilter к объекту.

    Args:
        obj (object): Объект, к которому применяются фильтры (например, event или message).
        *filters (MagicFilter): Один или несколько выражений MagicFilter.

    Returns:
        bool: True, если все фильтры возвращают True, иначе False.
    """

    try:
        return all(f.resolve(obj) for f in filters)
    except Exception:
        return False

`callback_payload` ¶

`CallbackPayload` ¶

Bases: BaseModel

Базовый класс для сериализации/десериализации callback payload.

Атрибуты

prefix (str): Префикс для payload (используется при pack/unpack) (по умолчанию название класса). separator (str): Разделитель между значениями (по умолчанию '|').

`__init_subclass__(**kwargs)` ¶

Автоматически проставляет prefix и separator при наследовании.

Source code in maxapi/filters/callback_payload.py

def __init_subclass__(cls, **kwargs: Any) -> None:
    """
    Автоматически проставляет prefix и separator при наследовании.
    """

    cls.prefix = kwargs.get("prefix", str(cls.__name__))
    cls.separator = kwargs.get("separator", "|")

`pack()` ¶

Собирает данные payload в строку для передачи в callback payload.

Raises:

Type	Description
`ValueError`	Если в значении встречается разделитель или payload слишком длинный.

Returns:

Name	Type	Description
`str`	`str`	Сериализованный payload.

Source code in maxapi/filters/callback_payload.py

def pack(self) -> str:
    """
    Собирает данные payload в строку для передачи в callback payload.

    Raises:
        ValueError: Если в значении встречается разделитель или payload слишком длинный.

    Returns:
        str: Сериализованный payload.
    """

    values = [self.prefix]

    for name in self.attrs():
        value = getattr(self, name)
        str_value = "" if value is None else str(value)
        if self.separator in str_value:
            raise ValueError(
                f'Символ разделителя "{self.separator}" не должен встречаться в значении поля {name}'
            )

        values.append(str_value)

    data = self.separator.join(values)

    if len(data.encode()) > PAYLOAD_MAX:
        raise ValueError(
            f"Payload слишком длинный! Максимум: {PAYLOAD_MAX} байт"
        )

    return data

`unpack(data)` `classmethod` ¶

Десериализует payload из строки.

Parameters:

Name	Type	Description	Default
`data`	`str`	Строка payload (из callback payload).	required

Raises:

Type	Description
`ValueError`	Некорректный prefix или количество аргументов.

Returns:

Name	Type	Description
`CallbackPayload`	`'CallbackPayload'`	Экземпляр payload с заполненными полями.

Source code in maxapi/filters/callback_payload.py

@classmethod
def unpack(cls, data: str) -> "CallbackPayload":
    """
    Десериализует payload из строки.

    Args:
        data (str): Строка payload (из callback payload).

    Raises:
        ValueError: Некорректный prefix или количество аргументов.

    Returns:
        CallbackPayload: Экземпляр payload с заполненными полями.
    """

    parts = data.split(cls.separator)

    if not parts[0] == cls.prefix:
        raise ValueError("Некорректный prefix")

    field_names = cls.attrs()

    if not len(parts) - 1 == len(field_names):
        raise ValueError(
            f"Ожидалось {len(field_names)} аргументов, получено {len(parts) - 1}"
        )

    kwargs = dict(zip(field_names, parts[1:]))
    return cls(**kwargs)

`attrs()` `classmethod` ¶

Возвращает список полей для сериализации/десериализации (исключая prefix и separator).

Returns:

Type	Description
`List[str]`	List[str]: Имена полей модели.

Source code in maxapi/filters/callback_payload.py

@classmethod
def attrs(cls) -> List[str]:
    """
    Возвращает список полей для сериализации/десериализации (исключая prefix и separator).

    Returns:
        List[str]: Имена полей модели.
    """

    return [
        k
        for k in cls.model_fields.keys()
        if k not in ("prefix", "separator")
    ]

`filter(rule=None)` `classmethod` ¶

Создаёт PayloadFilter для фильтрации callback-ивентов по payload.

Parameters:

Name	Type	Description	Default
`rule`	`Optional[MagicFilter]`	Фильтр на payload.	`None`

Returns:

Name	Type	Description
`PayloadFilter`	`PayloadFilter`	Экземпляр фильтра для хэндлера.

Source code in maxapi/filters/callback_payload.py

@classmethod
def filter(cls, rule: Optional[MagicFilter] = None) -> PayloadFilter:
    """
    Создаёт PayloadFilter для фильтрации callback-ивентов по payload.

    Args:
        rule (Optional[MagicFilter]): Фильтр на payload.

    Returns:
        PayloadFilter: Экземпляр фильтра для хэндлера.
    """

    return PayloadFilter(model=cls, rule=rule)

`PayloadFilter(model, rule)` ¶

Bases: BaseFilter

Фильтр для MessageCallback по payload.

Parameters:

Name	Type	Description	Default
`model`	`Type[CallbackPayload]`	Класс payload для распаковки.	required
`rule`	`Optional[MagicFilter]`	Фильтр (условие) для payload.	required

Source code in maxapi/filters/callback_payload.py

def __init__(
    self, model: Type[CallbackPayload], rule: Optional[MagicFilter]
):
    """
    Args:
        model (Type[CallbackPayload]): Класс payload для распаковки.
        rule (Optional[MagicFilter]): Фильтр (условие) для payload.
    """

    self.model = model
    self.rule = rule

`call(event)` `async` ¶

Проверяет event на MessageCallback и применяет фильтр к payload.

Parameters:

Name	Type	Description	Default
`event`	`UpdateUnion`	Обновление/событие.	required

Returns:

Type	Description
`Union[Dict[str, Any], bool]`	dict \| bool: dict с payload при совпадении, иначе False.

Source code in maxapi/filters/callback_payload.py

async def __call__(
    self, event: UpdateUnion
) -> Union[Dict[str, Any], bool]:
    """
    Проверяет event на MessageCallback и применяет фильтр к payload.

    Args:
        event (UpdateUnion): Обновление/событие.

    Returns:
        dict | bool: dict с payload при совпадении, иначе False.
    """

    if not isinstance(event, MessageCallback):
        return False

    if not event.callback.payload:
        return False

    try:
        payload = self.model.unpack(event.callback.payload)
    except Exception:
        return False

    if not self.rule or self.rule.resolve(payload):
        return {"payload": payload}

    return False

`command` ¶

`CommandsInfo(commands, info=None)` `dataclass` ¶

Датакласс информации о командах

Attributes:

Name	Type	Description
`commands`	`List[str]`	Список команд
`info`	`Optional[str]`	Информация о их предназначениях

`Command(commands, prefix='/', check_case=False, ignore_symbol_at_sign=False, only_with_bot_username=False)` ¶

Bases: BaseFilter

Фильтр сообщений на соответствие команде.

Parameters:

Name	Type	Description	Default
`commands`	`str \| List[str]`	Ожидаемая команда или список команд без префикса.	required
`prefix`	`str`	Префикс команды (по умолчанию '/').	`'/'`
`check_case`	`bool`	Учитывать регистр при сравнении (по умолчанию False).	`False`
`ignore_symbol_at_sign`	`bool`	Учитывать символ "@" при отправке команды с упоминанием бота (по умолчанию False).	`False`
`only_with_bot_username`	`bool`	Обязательно упоминать бота при отправке команды (по умолчанию False).	`False`

Инициализация фильтра команд.

Source code in maxapi/filters/command.py

def __init__(
    self,
    commands: str | List[str],
    prefix: str = "/",
    check_case: bool = False,
    ignore_symbol_at_sign: bool = False,
    only_with_bot_username: bool = False,
):
    """
    Инициализация фильтра команд.
    """

    if isinstance(commands, str):
        self.commands = [commands]
    else:
        self.commands = commands

    self.prefix = prefix
    self.check_case = check_case
    self.ignore_symbol_at_sign = ignore_symbol_at_sign
    self.only_with_bot_username = only_with_bot_username

    if not check_case:
        self.commands = [cmd.lower() for cmd in self.commands]

`parse_command(text, bot_username)` ¶

Извлекает команду из текста.

Parameters:

Name	Type	Description	Default
`text`	`str`	Текст сообщения.	required
`bot_username`	`str`	Имя пользователя бота.	required

Returns:

Type	Description
`str`	Tuple[str, List[str]]: Кортеж из команды без префикса и списка аргументов,
`List[str]`	либо ('', []) если команда не найдена или текст не соответствует формату.

Source code in maxapi/filters/command.py

def parse_command(
    self, text: str, bot_username: str
) -> Tuple[str, List[str]]:
    """
    Извлекает команду из текста.

    Args:
        text (str): Текст сообщения.
        bot_username (str): Имя пользователя бота.

    Returns:
        Tuple[str, List[str]]: Кортеж из команды без префикса и списка аргументов,
        либо ('', []) если команда не найдена или текст не соответствует формату.
    """

    if not text.strip():
        return "", []

    args = text.split()

    if not args:
        return "", []

    first = args[0]

    if self.ignore_symbol_at_sign:
        if first == bot_username:
            first = "@" + first

    if first.startswith("@"):
        if len(args) < 2:
            return "", []

        if not first[1:] == bot_username:
            return "", []

        command_part = args[1]

        if not command_part.startswith(self.prefix):
            return "", []

        command = command_part[len(self.prefix) :]
        arguments = args[2:]

    else:
        if self.only_with_bot_username:
            return "", []

        command_part = first

        if not command_part.startswith(self.prefix):
            return "", []

        command = command_part[len(self.prefix) :]
        arguments = args[1:]

    return command, arguments

`call(event)` `async` ¶

Проверяет, соответствует ли сообщение заданной(ым) команде(ам).

Parameters:

Name	Type	Description	Default
`event`	`MessageCreated`	Событие сообщения.	required

Returns:

Type	Description
`Union[Dict[str, List[str]], bool]`	dict \| bool: dict с аргументами команды при совпадении, иначе False.

Source code in maxapi/filters/command.py

async def __call__(
    self, event: UpdateUnion
) -> Union[Dict[str, List[str]], bool]:
    """
    Проверяет, соответствует ли сообщение заданной(ым) команде(ам).

    Args:
        event (MessageCreated): Событие сообщения.

    Returns:
        dict | bool: dict с аргументами команды при совпадении, иначе False.
    """

    if not isinstance(event, MessageCreated):
        return False

    # body может быть None — защитимся от обращения
    body = event.message.body
    if body is None:
        return False

    text = body.text
    if not text:
        return False

    # временно
    bot_me = event._ensure_bot().me
    bot_username = ""
    if bot_me:
        bot_username = bot_me.username or ""

    parsed_command, args = self.parse_command(text, bot_username)
    if not parsed_command:
        return False

    if not self.check_case:
        if parsed_command.lower() in [
            commands.lower() for commands in self.commands
        ]:
            return {"args": args}
        else:
            return False

    if parsed_command in self.commands:
        return {"args": args}

    return False

`CommandStart(prefix='/', check_case=False, ignore_symbol_at_sign=False, only_with_bot_username=False)` ¶

Bases: Command

Фильтр для команды /start.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	Префикс команды (по умолчанию '/').	`'/'`
`check_case`	`bool`	Учитывать регистр (по умолчанию False)	`False`
`ignore_symbol_at_sign`	`bool`	Учитывать символ "@" при отправке команды с упоминанием бота (по умолчанию False).	`False`
`only_with_bot_username`	`bool`	Обязательно упоминать бота при отправке команды (по умолчанию False)..	`False`

Source code in maxapi/filters/command.py

def __init__(
    self,
    prefix: str = "/",
    check_case: bool = False,
    ignore_symbol_at_sign: bool = False,
    only_with_bot_username: bool = False,
) -> None:
    super().__init__(
        "start",
        prefix,
        check_case,
        ignore_symbol_at_sign,
        only_with_bot_username,
    )

`filter` ¶

`BaseFilter` ¶

Базовый класс для фильтров.

Определяет интерфейс фильтрации событий. Потомки должны переопределять метод call.

Methods:

Name	Description
`__call__`	Асинхронная проверка события на соответствие фильтру.

`handler` ¶

`Handler(*args, func_event, update_type, **kwargs)` ¶

Обработчик события.

Связывает функцию-обработчик с типом события, состояниями и фильтрами.

Создаёт обработчик события.

Parameters:

Name	Type	Description	Default
`*args`	`Any`	Список фильтров (MagicFilter, State, Command, BaseFilter, BaseMiddleware).	`()`
`func_event`	`Callable`	Функция-обработчик.	required
`update_type`	`UpdateType`	Тип обновления.	required
`**kwargs`	`Any`	Дополнительные параметры.	`{}`

Source code in maxapi/filters/handler.py

def __init__(
    self,
    *args: Any,
    func_event: Callable,
    update_type: UpdateType,
    **kwargs: Any,
):
    """
    Создаёт обработчик события.

    Args:
        *args (Any): Список фильтров (MagicFilter, State, Command, BaseFilter, BaseMiddleware).
        func_event (Callable): Функция-обработчик.
        update_type (UpdateType): Тип обновления.
        **kwargs (Any): Дополнительные параметры.
    """

    self.func_event: Callable = func_event
    self.update_type: UpdateType = update_type
    self.filters: Optional[List[MagicFilter]] = []
    self.base_filters: Optional[List[BaseFilter]] = []
    self.states: Optional[List[State | None]] = []
    self.middlewares: List[BaseMiddleware] = []

    for arg in args:
        if isinstance(arg, MagicFilter):
            self.filters.append(arg)
        elif isinstance(arg, State) or arg is None:
            self.states.append(arg)
        elif isinstance(arg, BaseMiddleware):
            self.middlewares.append(arg)
        elif isinstance(arg, BaseFilter):
            self.base_filters.append(arg)
        else:
            logger_dp.info(
                f"Неизвестный фильтр `{arg}` при регистрации `{func_event.__name__}`"
            )

`middleware` ¶

`BaseMiddleware` ¶

Базовый класс для мидлварей.

Используется для обработки события до и после вызова хендлера.

`call(handler, event_object, data)` `async` ¶

Вызывает хендлер с переданным событием и данными.

Parameters:

Name	Type	Description	Default
`handler`	`Callable`	Хендлер события.	required
`event_object`	`Any`	Событие.	required
`data`	`dict`	Дополнительные данные.	required

Returns:

Name	Type	Description
`Any`	`Any`	Результат работы хендлера.

Source code in maxapi/filters/middleware.py

async def __call__(
    self,
    handler: Callable[[Any, dict[str, Any]], Awaitable[Any]],
    event_object: Any,
    data: dict[str, Any],
) -> Any:
    """
    Вызывает хендлер с переданным событием и данными.

    Args:
        handler (Callable): Хендлер события.
        event_object (Any): Событие.
        data (dict): Дополнительные данные.

    Returns:
        Any: Результат работы хендлера.
    """

    return await handler(event_object, data)

Package: maxapi.filters¶

maxapi.filters ¶

BaseFilter ¶

filter_attrs(obj, *filters) ¶

callback_payload ¶

CallbackPayload ¶

__init_subclass__(**kwargs) ¶

pack() ¶

unpack(data) classmethod ¶

attrs() classmethod ¶

filter(rule=None) classmethod ¶

PayloadFilter(model, rule) ¶

__call__(event) async ¶

command ¶

CommandsInfo(commands, info=None) dataclass ¶

Command(commands, prefix='/', check_case=False, ignore_symbol_at_sign=False, only_with_bot_username=False) ¶

parse_command(text, bot_username) ¶

__call__(event) async ¶

CommandStart(prefix='/', check_case=False, ignore_symbol_at_sign=False, only_with_bot_username=False) ¶

filter ¶

BaseFilter ¶

handler ¶

Handler(*args, func_event, update_type, **kwargs) ¶

middleware ¶

BaseMiddleware ¶

__call__(handler, event_object, data) async ¶

`maxapi.filters` ¶

`BaseFilter` ¶

`filter_attrs(obj, *filters)` ¶

`callback_payload` ¶

`CallbackPayload` ¶

`__init_subclass__(**kwargs)` ¶

`pack()` ¶

`unpack(data)` `classmethod` ¶

`attrs()` `classmethod` ¶

`filter(rule=None)` `classmethod` ¶

`PayloadFilter(model, rule)` ¶

`call(event)` `async` ¶

`command` ¶

`CommandsInfo(commands, info=None)` `dataclass` ¶

`Command(commands, prefix='/', check_case=False, ignore_symbol_at_sign=False, only_with_bot_username=False)` ¶

`parse_command(text, bot_username)` ¶

`call(event)` `async` ¶

`CommandStart(prefix='/', check_case=False, ignore_symbol_at_sign=False, only_with_bot_username=False)` ¶

`filter` ¶

`BaseFilter` ¶

`handler` ¶

`Handler(*args, func_event, update_type, **kwargs)` ¶

`middleware` ¶

`BaseMiddleware` ¶

`call(handler, event_object, data)` `async` ¶