--- metadata: - name: generator content: Diplodoc Platform v5.63.3 alternate: - https://ytsaurus.tech/docs/en/flow/python/examples/retryable_async_request.md - https://ytsaurus.tech/docs/ru/flow/python/examples/retryable_async_request.md - href: https://ytsaurus.tech/docs/ru/flow/python/examples/retryable_async_request.md type: text/markdown title: Markdown version - href: https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 rel: describedby --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 # Retryable Async Request в YTsaurus Flow (Python) Расширение примера [AsyncRequest](https://ytsaurus.tech/docs/ru/flow/python/examples/async_request.md): обработчик запросов поддерживает автоматические повторные попытки через [таймеры](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#timer). При неудаче запрос сохраняется во внутреннем [стейте](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state) и повторяется через фиксированный интервал, вплоть до `MAX_RETRIES` попыток. [Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/python/retryable_async_request) ## Структура Пайплайн состоит из двух компьютейшенов: 1. **`state`** (`StateKeeperFunction`) — stateful-компьютейшен: маршрутизирует события в запросы и накапливает суммарную длину ответов во внешнем стейте (аналогично `async_request`). 2. **`processor`** (`RequestProcessorFunction`) — stateful-компьютейшен с логикой повторов: - При первой попытке сохраняет запрос во внутреннем стейте и пытается его обработать. - Если обработка не удалась — увеличивает счётчик неудач, сохраняет стейт и устанавливает таймер (`DELAY_SECONDS = 5`). - По таймеру повторяет попытку (до `MAX_RETRIES = 3`). - При успехе — отправляет ответ и очищает стейт. ## `request_processor_function.py` Ключевой файл примера: реализует логику повторных попыток через `on_timer` и вспомогательный метод `_try_or_retry`. ```python class RequestProcessorFunction(RowFunction): """Retries requests up to MAX_RETRIES times; emits response on success.""" def on_message(self, message, output, ctx): request_id = message.payload["request_id"] data = { "request_id": request_id, "key": message.payload["key"], "request": message.payload["request"], "failed_attempts": 0, } state = ctx.state("request-state", message) self._try_or_retry(state, data, output, ctx) def on_timer(self, timer, output, ctx): state = ctx.state("request-state", timer) data = state.get_or_default(None) if data is not None: self._try_or_retry(state, data, output, ctx) def _try_or_retry(self, state, data, output, ctx): request_id = data["request_id"] failed_attempts = data["failed_attempts"] if not _is_request_successful(request_id, failed_attempts): data["failed_attempts"] += 1 state.set(data) output.add_timer(int(time.time()) + DELAY_SECONDS) log.debug( "Failed request (RequestId: %s, FailedAttempts: %s)", request_id, data["failed_attempts"], ) else: key = data["key"] length = len(data["request"]) builder = ctx.message_builder("response") builder.set("request_id", request_id) builder.set("key", key) builder.set("length", length) output.add_message(builder.finish()) state.clear() log.debug( "Processed request (RequestId: %s, FailedAttempts: %s)", request_id, failed_attempts, ) ``` ## `state_keeper_function.py` Маршрутизация стримов и накопление результатов во внешнем стейте (идентично `async_request`). ```python class StateKeeperFunction(RowFunction): """Routes event messages to requests and accumulates response lengths in external state.""" def on_message(self, message, output, ctx): stream_id = message.stream_id if stream_id == "event": key = message.payload["key"] data = message.payload["data"] request_id = random.getrandbits(64) builder = ctx.message_builder("request") builder.set("request_id", request_id) builder.set("key", key) builder.set("request", data) output.add_message(builder.finish()) log.debug("Send request (RequestId: %s)", request_id) elif stream_id == "response": request_id = message.payload["request_id"] length = message.payload["length"] log.debug("Received response (RequestId: %s)", request_id) state = ctx.external_state("/state", message) current = state.get("total_length") total_length = current if current is not None else 0 total_length += length builder = state.to_builder() builder.set("total_length", total_length) state.set(builder.finish()) else: raise ValueError(f"Unexpected stream_id: {stream_id}") ``` ## `__main__.py` Точка входа: структура идентична `async_request`, отличается только импортируемыми классами. ```python def main(): log.info("Starting companion execution") pipeline = Pipeline() pipeline.add("state", StateKeeperFunction()) pipeline.add("processor", RequestProcessorFunction()) log.info("Starting pipeline...") pipeline.run() log.info("Pipeline completed") ``` ## Ключевые паттерны - **Retry через таймеры**: `output.add_timer(int(time.time()) + DELAY_SECONDS)` откладывает повторную попытку; `on_timer` читает стейт и повторяет `_try_or_retry` — стандартный паттерн реализации повторов в Flow без внешних очередей. - **Счётчик попыток в стейте**: поле `failed_attempts` сохраняется вместе с данными запроса в `ctx.state("request-state", message)`, что гарантирует корректность при рестарте. - **Детерминированная симуляция сбоев**: `_is_request_succeed(request_id, failed_attempts)` моделирует нестабильный внешний сервис; в реальном коде заменяется вызовом HTTP-клиента. - **Очистка стейта при успехе**: `state.clear()` после успешной обработки предотвращает повторное срабатывание таймера на устаревших данных. - **Разделение ответственности**: `StateKeeperFunction` не знает о повторах — вся логика retry инкапсулирована в `RequestProcessorFunction`, что упрощает замену стратегии повторов. ## См. также - [Быстрый старт (Python)](https://ytsaurus.tech/docs/ru/flow/python/getting-started.md) - [Computation (Python)](https://ytsaurus.tech/docs/ru/flow/python/computation.md) - [Таймеры](https://ytsaurus.tech/docs/ru/flow/concepts/timers.md) - [Stateful processing](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md) - [AsyncRequest (Python)](https://ytsaurus.tech/docs/ru/flow/python/examples/async_request.md) - [Аналогичный пример на C++](https://ytsaurus.tech/docs/ru/flow/cpp/examples/retryable_async_request.md) - [Аналогичный пример на Java](https://ytsaurus.tech/docs/ru/flow/java/examples/retryable_async_request.md)