---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/python/examples/async_request.md
  - https://ytsaurus.tech/docs/ru/flow/python/examples/async_request.md
  - href: https://ytsaurus.tech/docs/ru/flow/python/examples/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

<!-- source: ru/_includes/flow/python/examples/async_request.md -->
# Async Request в YTsaurus Flow (Python)

Пример двухкомпонентного [пайплайна](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#pipeline), реализующего асинхронную обработку запросов: один [компьютейшен](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation) маршрутизирует события в запросы и накапливает ответы, другой — выполняет вычисление без [стейта](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state). Python-реализация аналогичного [C++ примера](https://ytsaurus.tech/docs/ru/flow/cpp/examples/async_request.md).

[Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/python/async_request)

## Структура

Пайплайн состоит из двух компьютейшенов:

1. **`state`** (`StateKeeperFunction`) — stateful-компьютейшен, который:
   - Принимает события из стрима `event` и порождает запросы в стрим `request` с уникальным `request_id`.
   - Принимает ответы из стрима `response` и накапливает суммарную длину (`total_length`) во внешнем [стейте](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state).

2. **`processor`** (`RequestProcessorFunction`) — stateless-компьютейшен, который принимает запросы из стрима `request` и немедленно возвращает ответ (длину строки запроса) в стрим `response`.

Цикл `event → request → response → state` замыкается между двумя компьютейшенами.

## `state_keeper_function.py`

Маршрутизация входящих стримов (`event` / `response`) и работа с внешним стейтом.

```python
class StateKeeperFunction(RowFunction):
    """Sends requests for events; accumulates total_length for responses."""

    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":
            length = message.payload["length"]
            request_id = message.payload["request_id"]
            log.debug("Received response (RequestId: %s)", request_id)

            state = ctx.external_state("/state", message)
            current = state.get("total_length")
            total = (current if current is not None else 0) + length
            builder = state.to_builder()
            builder.set("total_length", total)
            state.set(builder.finish())
        else:
            raise ValueError(f"Unknown stream_id: {stream_id}")

```

## `request_processor_function.py`

Stateless-обработчик запросов: вычисляет длину строки запроса и возвращает ответ.

```python
class RequestProcessorFunction(RowFunction):
    """Stateless: returns length of request string as response."""

    def on_message(self, message, output, ctx):
        request_id = message.payload["request_id"]
        key = message.payload["key"]
        request = message.payload["request"]
        length = len(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())
        log.debug("Processed request (RequestId: %s)", request_id)
```

## `__main__.py`

Точка входа: создание пайплайна и регистрация обоих компьютейшенов.

```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")
```

## Ключевые паттерны

- **Маршрутизация по `stream_id`**: ветвление `if stream_id == "event" / "response"` позволяет одному компьютейшену обрабатывать несколько входных стримов с разной логикой.
- **Генерация уникального `request_id`**: `random.getrandbits(64)` обеспечивает корреляцию между запросом и ответом в асинхронном цикле.
- **External state** через `ctx.external_state("/state", message)`: паттерн `to_builder()` / `set()` для накопительного обновления внешнего стейта.
- **Stateless-компьютейшен**: `RequestProcessorFunction` не использует стейт — чистое преобразование запроса в ответ, что позволяет масштабировать его независимо.
- **Двухкомпонентный пайплайн**: `pipeline.add("state", ...)` и `pipeline.add("processor", ...)` регистрируют компьютейшены, стримы между которыми описываются в [спеке](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#spec-and-dynamic-spec).
<!-- endsource: ru/_includes/flow/python/examples/async_request.md -->

<!-- source: ru/_includes/flow/python/examples/async_request_also.md -->
## См. также

- [Быстрый старт (Python)](https://ytsaurus.tech/docs/ru/flow/python/getting-started.md)
- [Computation (Python)](https://ytsaurus.tech/docs/ru/flow/python/computation.md)
- [Stateful processing](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md)
- [Аналогичный пример на C++](https://ytsaurus.tech/docs/ru/flow/cpp/examples/async_request.md)
- [Аналогичный пример на Java](https://ytsaurus.tech/docs/ru/flow/java/examples/async_request.md)
<!-- endsource: ru/_includes/flow/python/examples/async_request_also.md -->