GettedUpdates

warn_unprocessable_event(event)

Предупредить в лог о событии, которое не удалось разобрать.

Различает три случая: известную проблему на стороне MAX (message_created без тела сообщения — например, голосовое сообщение боту), известный тип события с некорректным содержимым и неизвестный тип события. В лог всегда попадает payload события, а в сообщении указано, куда обращаться.

Parameters:

Name	Type	Description	Default
event	dict[str, Any]	Словарь события из ответа API или вебхука.	required

Source code in maxapi/methods/types/getted_updates.py

def warn_unprocessable_event(event: dict[str, Any]) -> None:
    """Предупредить в лог о событии, которое не удалось разобрать.

    Различает три случая: известную проблему на стороне MAX
    (message_created без тела сообщения — например, голосовое
    сообщение боту), известный тип события с некорректным
    содержимым и неизвестный тип события. В лог всегда попадает
    payload события, а в сообщении указано, куда обращаться.

    Args:
        event: Словарь события из ответа API или вебхука.
    """

    update_type = event.get("update_type")
    event_json = _dump_event_json(event)

    if update_type == UpdateType.MESSAGE_CREATED and "message" not in event:
        logger_dp.warning(
            UNSUPPORTED_MESSAGE_UPDATE_DISCLAIMER.format(
                event_json=event_json,
            )
        )
    elif update_type in _KNOWN_UPDATE_TYPES:
        logger_dp.warning(
            MALFORMED_UPDATE_DISCLAIMER.format(
                update_type=update_type,
                event_json=event_json,
            )
        )
    else:
        logger_dp.warning(
            UNKNOWN_UPDATE_DISCLAIMER.format(
                update_type=update_type,
                event_json=event_json,
            )
        )

get_update_model(event, bot) async

Конвертировать словарь с событием в модель обновления.

При любой ошибке валидации (неизвестный тип события или известный тип с некорректным содержимым) возвращает None, чтобы не ломать процесс получения обновлений. Классификацию причины и предупреждение в лог выполняет вызывающий код через :func:warn_unprocessable_event.

Parameters:

Name	Type	Description	Default
event	dict[str, Any]	Словарь события из ответа API или вебхука.	required
bot	Bot	Экземпляр бота.	required

Returns:

Type	Description
UpdateUnion | None	Модель события или None, если событие не распознано.

Source code in maxapi/methods/types/getted_updates.py

async def get_update_model(
    event: dict[str, Any], bot: "Bot"
) -> UpdateUnion | None:
    """Конвертировать словарь с событием в модель обновления.

    При любой ошибке валидации (неизвестный тип события или
    известный тип с некорректным содержимым) возвращает ``None``,
    чтобы не ломать процесс получения обновлений. Классификацию
    причины и предупреждение в лог выполняет вызывающий код через
    :func:`warn_unprocessable_event`.

    Args:
        event: Словарь события из ответа API или вебхука.
        bot: Экземпляр бота.

    Returns:
        Модель события или ``None``, если событие не распознано.
    """

    try:
        event_object = UpdateUnionAdapter.validate_python(event)
    except ValueError:
        # Пришло новое событие, которое данная библиотека пока
        # не умеет обрабатывать. Возвращаем None, чтобы обработать это
        # в вызывающем коде и не ломать процесс получения обновлений
        return None

    return await enrich_event(event_object=event_object, bot=bot)

process_update_request(events, bot) async

Конвертировать словарь с обновлениями в список моделей.

Нераспознанные события пропускаются с предупреждением в лог. Ошибка разбора или обогащения одного события не прерывает обработку остальной пачки: проблемное событие пропускается с записью в лог.

Parameters:

Name	Type	Description	Default
events	dict[str, Any]	Ответ API метода GET /updates.	required
bot	Bot	Экземпляр бота.	required

Returns:

Type	Description
list[UpdateUnion]	Список разобранных событий.

Source code in maxapi/methods/types/getted_updates.py

async def process_update_request(
    events: dict[str, Any],
    bot: "Bot",
) -> list[UpdateUnion]:
    """Конвертировать словарь с обновлениями в список моделей.

    Нераспознанные события пропускаются с предупреждением в лог.
    Ошибка разбора или обогащения одного события не прерывает
    обработку остальной пачки: проблемное событие пропускается
    с записью в лог.

    Args:
        events: Ответ API метода GET /updates.
        bot: Экземпляр бота.

    Returns:
        Список разобранных событий.
    """

    events_models = []

    for event in events.get("updates") or []:
        try:
            event_model = await get_update_model(event, bot)
            if event_model is None:
                # Классификация тоже внутри try: она обращается к
                # event как к словарю, а элемент пачки может им не
                # быть (например, null) — тогда пропускаем событие,
                # а не роняем разбор всей пачки.
                warn_unprocessable_event(event)
                continue
        except Exception as exc:
            logger_dp.exception(
                "Событие пропущено из-за ошибки разбора: %r | %s",
                exc,
                _dump_event_json(event),
            )
            continue

        events_models.append(event_model)

    return events_models

process_update_webhook(event_json, bot) async

Конвертировать JSON события вебхука в модель обновления.

Parameters:

Name	Type	Description	Default
event_json	dict[str, Any]	Тело вебхук-запроса.	required
bot	Bot	Экземпляр бота.	required

Returns:

Type	Description
UpdateUnion | None	Модель события или None, если событие не распознано.

Source code in maxapi/methods/types/getted_updates.py

async def process_update_webhook(
    event_json: dict[str, Any], bot: "Bot"
) -> UpdateUnion | None:
    """Конвертировать JSON события вебхука в модель обновления.

    Args:
        event_json: Тело вебхук-запроса.
        bot: Экземпляр бота.

    Returns:
        Модель события или ``None``, если событие не распознано.
    """

    return await get_update_model(bot=bot, event=event_json)