CallbackPayload¶

`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

CallbackPayload¶