> **Источник:** https://python-all.ru/3.9/library/asyncio-stream.html
>
> «Документация Python на русском» – неофициальный перевод официальной документации Python: версии от 2.6 до 3.16, полнотекстовый поиск, английский оригинал рядом с переводом. Эта Markdown-версия страницы предназначена для работы с LLM: вставьте её в ChatGPT, Claude или Cursor.

---

# Потоки данных

**Исходный код:** [Lib/asyncio/streams.py](https://python-all.ru/src/3.9/Lib/asyncio/streams.py)

---

Потоки данных – это высокоуровневые примитивы, готовые к использованию с async/await, для работы с сетевыми соединениями. Потоки данных позволяют отправлять и получать данные без использования колбэков или низкоуровневых протоколов и транспортов.

Ниже приведён пример TCP-эхо-клиента, написанного с использованием asyncio потоков данных:

```python
import asyncio

async def tcp_echo_client(message):
    reader, writer = await asyncio.open_connection(
        '127.0.0.1', 8888)

    print(f'Send: {message!r}')
    writer.write(message.encode())
    await writer.drain()

    data = await reader.read(100)
    print(f'Received: {data.decode()!r}')

    print('Close the connection')
    writer.close()
    await writer.wait_closed()

asyncio.run(tcp_echo_client('Hello World!'))
```

Смотрите также раздел [Примеры](https://python-all.ru/3.9/library/asyncio-stream.html#examples) ниже.

Функции потоков данных

Следующие функции asyncio верхнего уровня можно использовать для создания и работы с потоками данных:

#### `coroutine asyncio.open_connection(host=None, port=None, *, loop=None, limit=None, ssl=None, family=0, proto=0, flags=0, sock=None, local_addr=None, server_hostname=None, ssl_handshake_timeout=None, happy_eyeballs_delay=None, interleave=None)`

Устанавливает сетевое соединение и возвращает пару объектов `(reader, writer)`.

Возвращаемые объекты *reader* и *writer* являются экземплярами классов [`StreamReader`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.StreamReader) и [`StreamWriter`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.StreamWriter).

Аргумент *loop* необязателен и всегда определяется автоматически при ожидании этой функции из корутины.

Параметр *limit* определяет ограничение размера буфера, используемого возвращаемым экземпляром [`StreamReader`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.StreamReader). По умолчанию *limit* установлен в 64 КиБ.

Остальные аргументы передаются напрямую в [`loop.create_connection()`](https://python-all.ru/3.9/library/asyncio-eventloop.html#asyncio.loop.create_connection).

Новое в версии 3.7: параметр *ssl\_handshake\_timeout*.

Новое в версии 3.8: Добавлены параметры *happy\_eyeballs\_delay* и *interleave*.

#### `coroutine asyncio.start_server(client_connected_cb, host=None, port=None, *, loop=None, limit=None, family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE, sock=None, backlog=100, ssl=None, reuse_address=None, reuse_port=None, ssl_handshake_timeout=None, start_serving=True)`

Запускает сокет-сервер.

Колбэк *client\_connected\_cb* вызывается при каждом установлении нового подключения клиента. Он получает пару `(reader, writer)` в качестве двух аргументов – экземпляры классов [`StreamReader`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.StreamReader) и [`StreamWriter`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.StreamWriter).

*client\_connected\_cb* может быть обычным вызываемым объектом или [корутинной функцией](https://python-all.ru/3.9/library/asyncio-task.html#coroutine); если это корутинная функция, она будет автоматически запланирована как [`Task`](https://python-all.ru/3.9/library/asyncio-task.html#asyncio.Task).

Аргумент *loop* необязателен и всегда определяется автоматически при ожидании этого метода из корутины.

Параметр *limit* определяет ограничение размера буфера, используемого возвращаемым экземпляром [`StreamReader`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.StreamReader). По умолчанию *limit* установлен в 64 КиБ.

Остальные аргументы передаются напрямую в [`loop.create_server()`](https://python-all.ru/3.9/library/asyncio-eventloop.html#asyncio.loop.create_server).

Новое в версии 3.7: параметры *ssl\_handshake\_timeout* и *start\_serving*.

Unix-сокеты

#### `coroutine asyncio.open_unix_connection(path=None, *, loop=None, limit=None, ssl=None, sock=None, server_hostname=None, ssl_handshake_timeout=None)`

Устанавливает соединение через Unix-сокет и возвращает пару `(reader, writer)`.

Аналогично [`open_connection()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.open_connection), но работает с Unix-сокетами.

См. также документацию [`loop.create_unix_connection()`](https://python-all.ru/3.9/library/asyncio-eventloop.html#asyncio.loop.create_unix_connection).

[Доступность](https://python-all.ru/3.9/library/intro.html#availability): Unix.

Новое в версии 3.7: параметр *ssl\_handshake\_timeout*.

Изменено в версии 3.7: Параметр *path* теперь может быть [объектом, подобным пути](https://python-all.ru/3.9/glossary.html#term-path-like-object)

#### `coroutine asyncio.start_unix_server(client_connected_cb, path=None, *, loop=None, limit=None, sock=None, backlog=100, ssl=None, ssl_handshake_timeout=None, start_serving=True)`

Запуск сервера Unix-сокетов.

Аналогично [`start_server()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.start_server), но работает с Unix-сокетами.

См. также документацию [`loop.create_unix_server()`](https://python-all.ru/3.9/library/asyncio-eventloop.html#asyncio.loop.create_unix_server).

[Доступность](https://python-all.ru/3.9/library/intro.html#availability): Unix.

Новое в версии 3.7: параметры *ssl\_handshake\_timeout* и *start\_serving*.

Изменено в версии 3.7: Параметр *path* теперь может быть [объектом, подобным пути](https://python-all.ru/3.9/glossary.html#term-path-like-object).

## StreamReader

#### `class asyncio.StreamReader`

Представляет объект чтения, предоставляющий API для чтения данных из потока ввода-вывода.

Не рекомендуется создавать объекты *StreamReader* напрямую; вместо этого используйте [`open_connection()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.open_connection) и [`start_server()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.start_server).

#### `coroutine read(n=-1)`

Читает до *n* байт. Если *n* не указан или равен `-1`, читает до EOF и возвращает все прочитанные байты.

Если получен EOF и внутренний буфер пуст, возвращает пустой объект `bytes`.

#### `coroutine readline()`

Читает одну строку, где «строка» – это последовательность байтов, заканчивающаяся `\n`.

Если получен EOF и `\n` не найден, метод возвращает частично прочитанные данные.

Если получен EOF и внутренний буфер пуст, возвращается пустой объект `bytes`.

#### `coroutine readexactly(n)`

Прочитать ровно *n* байт.

Возбуждается [`IncompleteReadError`](https://python-all.ru/3.9/library/asyncio-exceptions.html#asyncio.IncompleteReadError), если EOF достигнут до того, как прочитано *n* байт. Используйте атрибут [`IncompleteReadError.partial`](https://python-all.ru/3.9/library/asyncio-exceptions.html#asyncio.IncompleteReadError.partial), чтобы получить частично прочитанные данные.

#### `coroutine readuntil(separator=b'\n')`

Читает данные из потока до тех пор, пока не будет найден *separator*.

В случае успеха данные и разделитель удаляются из внутреннего буфера (потребляются). Возвращаемые данные включают разделитель в конце.

Если объем прочитанных данных превышает настроенный лимит потока, возбуждается исключение [`LimitOverrunError`](https://python-all.ru/3.9/library/asyncio-exceptions.html#asyncio.LimitOverrunError), а данные остаются во внутреннем буфере и могут быть прочитаны снова.

Если EOF достигнут до того, как найден полный разделитель, возбуждается исключение [`IncompleteReadError`](https://python-all.ru/3.9/library/asyncio-exceptions.html#asyncio.IncompleteReadError), и внутренний буфер сбрасывается. Атрибут [`IncompleteReadError.partial`](https://python-all.ru/3.9/library/asyncio-exceptions.html#asyncio.IncompleteReadError.partial) может содержать часть разделителя.

Новое в версии 3.5.2.

#### `at_eof()`

Возвращает `True`, если буфер пуст и был вызван `feed_eof()`.

## StreamWriter

#### `class asyncio.StreamWriter`

Представляет объект-писатель, предоставляющий API для записи данных в поток ввода-вывода.

Не рекомендуется создавать объекты *StreamWriter* напрямую; вместо этого используйте [`open_connection()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.open_connection) и [`start_server()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.start_server).

#### `write(data)`

Метод пытается немедленно записать *data* в нижележащий сокет. Если это не удаётся, данные помещаются во внутренний буфер записи до тех пор, пока их нельзя будет отправить.

Метод следует использовать вместе с методом `drain()`:

```python
stream.write(data)
await stream.drain()
```

#### `writelines(data)`

Метод немедленно записывает список (или любой итерируемый объект) байт в нижележащий сокет. Если это не удаётся, данные помещаются во внутренний буфер записи до тех пор, пока их нельзя будет отправить.

Метод следует использовать вместе с методом `drain()`:

```python
stream.writelines(lines)
await stream.drain()
```

#### `close()`

Метод закрывает поток и нижележащий сокет.

Метод следует использовать вместе с методом `wait_closed()`:

```python
stream.close()
await stream.wait_closed()
```

#### `can_write_eof()`

Возвращает `True`, если нижележащий транспорт поддерживает метод [`write_eof()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.StreamWriter.write_eof), и `False` в противном случае.

#### `write_eof()`

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

#### `transport`

Возвращает базовый транспорт asyncio.

#### `get_extra_info(name, default=None)`

Предоставляет доступ к дополнительной информации о транспорте; подробнее см. [`BaseTransport.get_extra_info()`](https://python-all.ru/3.9/library/asyncio-protocol.html#asyncio.BaseTransport.get_extra_info).

#### `coroutine drain()`

Ожидает, когда можно будет возобновить запись в поток данных. Пример:

```python
writer.write(data)
await writer.drain()
```

Это метод управления потоком, который взаимодействует с базовым буфером записи ввода-вывода. Когда размер буфера достигает верхней границы, *drain()* блокируется, пока размер буфера не уменьшится до нижней границы, после чего запись можно возобновить. Если ждать нечего, [`drain()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.StreamWriter.drain) возвращает управление немедленно.

#### `is_closing()`

Возвращает `True`, если поток данных закрыт или находится в процессе закрытия.

Добавлено в версии 3.7.

#### `coroutine wait_closed()`

Ожидает закрытия потока данных.

Должен вызываться после [`close()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.StreamWriter.close), чтобы дождаться закрытия нижележащего соединения.

Добавлено в версии 3.7.

## Примеры

### TCP-эхо-клиент с использованием потоков данных

TCP-эхо-клиент с использованием функции [`asyncio.open_connection()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.open_connection):

```python
import asyncio

async def tcp_echo_client(message):
    reader, writer = await asyncio.open_connection(
        '127.0.0.1', 8888)

    print(f'Send: {message!r}')
    writer.write(message.encode())

    data = await reader.read(100)
    print(f'Received: {data.decode()!r}')

    print('Close the connection')
    writer.close()

asyncio.run(tcp_echo_client('Hello World!'))
```

> **См. также**
>
> Пример [протокол TCP-эхо-клиента](https://python-all.ru/3.9/library/asyncio-protocol.html#asyncio-example-tcp-echo-client-protocol) использует низкоуровневый метод [`loop.create_connection()`](https://python-all.ru/3.9/library/asyncio-eventloop.html#asyncio.loop.create_connection).

### TCP-эхо-сервер с использованием потоков данных

TCP-эхо-сервер с использованием функции [`asyncio.start_server()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.start_server):

```python
import asyncio

async def handle_echo(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')

    print(f"Received {message!r} from {addr!r}")

    print(f"Send: {message!r}")
    writer.write(data)
    await writer.drain()

    print("Close the connection")
    writer.close()

async def main():
    server = await asyncio.start_server(
        handle_echo, '127.0.0.1', 8888)

    addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
    print(f'Serving on {addrs}')

    async with server:
        await server.serve_forever()

asyncio.run(main())
```

> **См. также**
>
> Пример [протокол TCP-эхо-сервера](https://python-all.ru/3.9/library/asyncio-protocol.html#asyncio-example-tcp-echo-server-protocol) использует метод [`loop.create_server()`](https://python-all.ru/3.9/library/asyncio-eventloop.html#asyncio.loop.create_server).

### Получение HTTP-заголовков

Простой пример получения HTTP-заголовков URL, переданного в командной строке:

```python
import asyncio
import urllib.parse
import sys

async def print_http_headers(url):
    url = urllib.parse.urlsplit(url)
    if url.scheme == 'https':
        reader, writer = await asyncio.open_connection(
            url.hostname, 443, ssl=True)
    else:
        reader, writer = await asyncio.open_connection(
            url.hostname, 80)

    query = (
        f"HEAD {url.path or '/'} HTTP/1.0\r\n"
        f"Host: {url.hostname}\r\n"
        f"\r\n"
    )

    writer.write(query.encode('latin-1'))
    while True:
        line = await reader.readline()
        if not line:
            break

        line = line.decode('latin1').rstrip()
        if line:
            print(f'HTTP header> {line}')

    # Игнорировать тело, закрыть сокет
    writer.close()

url = sys.argv[1]
asyncio.run(print_http_headers(url))
```

Использование:

```python
python example.py http://example.com/path/page.html
```

или с HTTPS:

```python
python example.py https://example.com/path/page.html
```

### Регистрация открытого сокета для ожидания данных с использованием потоков данных

Корутина, ожидающая получения данных сокетом с помощью функции [`open_connection()`](https://python-all.ru/3.9/library/asyncio-stream.html#asyncio.open_connection):

```python
import asyncio
import socket

async def wait_for_data():
    # Получить ссылку на текущий цикл событий, так как
    # мы хотим получить доступ к низкоуровневым API.
    loop = asyncio.get_running_loop()

    # Создать пару соединённых сокетов.
    rsock, wsock = socket.socketpair()

    # Зарегистрировать открытый сокет для ожидания данных.
    reader, writer = await asyncio.open_connection(sock=rsock)

    # Симулировать приём данных из сети
    loop.call_soon(wsock.send, 'abc'.encode())

    # Ожидать данные
    data = await reader.read(100)

    # Данные получены, всё готово: закрыть сокет
    print("Received:", data.decode())
    writer.close()

    # Закрыть второй сокет
    wsock.close()

asyncio.run(wait_for_data())
```

> **См. также**
>
> Пример [регистрация открытого сокета для ожидания данных с использованием протокола](https://python-all.ru/3.9/library/asyncio-protocol.html#asyncio-example-create-connection) использует низкоуровневый протокол и метод [`loop.create_connection()`](https://python-all.ru/3.9/library/asyncio-eventloop.html#asyncio.loop.create_connection).
>
> Пример [наблюдения за файловым дескриптором на предмет событий чтения](https://python-all.ru/3.9/library/asyncio-eventloop.html#asyncio-example-watch-fd) использует низкоуровневый метод [`loop.add_reader()`](https://python-all.ru/3.9/library/asyncio-eventloop.html#asyncio.loop.add_reader) для наблюдения за файловым дескриптором.