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

---

# 18.5.5. Потоки данных (API на основе корутин)

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

> **Примечание**
>
> Функции верхнего уровня этого модуля – это лишь удобные обёртки; в них нет ничего особенного, и если они делают не совсем то, что вам нужно, можно просто скопировать их код.

#### `coroutine asyncio.open_connection(host=None, port=None, *, loop=None, limit=None, **kwds)`

Обёртка для [`create_connection()`](https://python-all.ru/3.5/library/asyncio-eventloop.html#asyncio.AbstractEventLoop.create_connection), возвращающая пару (reader, writer).

Возвращаемый reader является экземпляром [`StreamReader`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamReader); writer – экземпляром [`StreamWriter`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamWriter).

Аргументы совпадают с обычными аргументами [`AbstractEventLoop.create_connection()`](https://python-all.ru/3.5/library/asyncio-eventloop.html#asyncio.AbstractEventLoop.create_connection), за исключением *protocol\_factory*; наиболее распространёнными являются позиционные аргументы host и port, за которыми могут следовать различные необязательные именованные аргументы.

Дополнительные необязательные именованные аргументы: *loop* (для указания используемого экземпляра цикла событий) и *limit* (для установки лимита буфера, передаваемого в [`StreamReader`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamReader)).

Эта функция – [корутина](https://python-all.ru/3.5/library/asyncio-task.html#coroutine).

#### `coroutine asyncio.start_server(client_connected_cb, host=None, port=None, *, loop=None, limit=None, **kwds)`

Запускает сокет-сервер с колбэком для каждого подключившегося клиента. Возвращаемое значение такое же, как у [`create_server()`](https://python-all.ru/3.5/library/asyncio-eventloop.html#asyncio.AbstractEventLoop.create_server).

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

Остальные аргументы – это все обычные аргументы [`create_server()`](https://python-all.ru/3.5/library/asyncio-eventloop.html#asyncio.AbstractEventLoop.create_server), за исключением *protocol\_factory*; наиболее распространёнными являются позиционные аргументы *host* и *port*, за которыми могут следовать различные необязательные именованные аргументы.

Дополнительные необязательные именованные аргументы: *loop* (для указания используемого экземпляра цикла событий) и *limit* (для установки лимита буфера, передаваемого в [`StreamReader`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamReader)).

Эта функция – [корутина](https://python-all.ru/3.5/library/asyncio-task.html#coroutine).

#### `coroutine asyncio.open_unix_connection(path=None, *, loop=None, limit=None, **kwds)`

Обёртка для [`create_unix_connection()`](https://python-all.ru/3.5/library/asyncio-eventloop.html#asyncio.AbstractEventLoop.create_unix_connection), возвращающая пару (reader, writer).

См. [`open_connection()`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.open_connection) для получения сведений о возвращаемом значении и других подробностях.

Эта функция – [корутина](https://python-all.ru/3.5/library/asyncio-task.html#coroutine).

Доступность: UNIX.

#### `coroutine asyncio.start_unix_server(client_connected_cb, path=None, *, loop=None, limit=None, **kwds)`

Запускает сервер UNIX Domain Socket с колбэком для каждого подключившегося клиента.

См. [`start_server()`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.start_server) для получения сведений о возвращаемом значении и других подробностях.

Эта функция – [корутина](https://python-all.ru/3.5/library/asyncio-task.html#coroutine).

Доступность: UNIX.

## 18.5.5.2. StreamReader

#### `class asyncio.StreamReader(limit=None, loop=None)`

This class is [не является потокобезопасным](https://python-all.ru/3.5/library/asyncio-dev.html#asyncio-multithreading).

#### `exception()`

Получить исключение.

#### `feed_eof()`

Подтверждает конец файла (EOF).

#### `feed_data(data)`

Подаёт байты *data* во внутренний буфер. Любые операции, ожидающие данные, будут возобновлены.

#### `set_exception(exc)`

Устанавливает исключение.

#### `set_transport(transport)`

Устанавливает транспорт.

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

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

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

Этот метод является [корутиной](https://python-all.ru/3.5/library/asyncio-task.html#coroutine).

#### `coroutine readline()`

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

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

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

Этот метод является [корутиной](https://python-all.ru/3.5/library/asyncio-task.html#coroutine).

#### `coroutine readexactly(n)`

Читает ровно *n* байт. Вызывает [`IncompleteReadError`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.IncompleteReadError), если конец потока достигнут до того, как можно прочитать *n*. Атрибут [`IncompleteReadError.partial`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.IncompleteReadError.partial) исключения содержит частично прочитанные байты.

Этот метод является [корутиной](https://python-all.ru/3.5/library/asyncio-task.html#coroutine).

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

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

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

Настроенный лимит потока используется для проверки результата. Лимит определяет максимальную длину возвращаемых данных, не включая разделитель.

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

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

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

#### `at_eof()`

Возвращает `True`, если буфер пуст и был вызван [`feed_eof()`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamReader.feed_eof).

## 18.5.5.3. StreamWriter

#### `class asyncio.StreamWriter(transport, protocol, reader, loop)`

Оборачивает Transport.

Этот класс предоставляет [`write()`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamWriter.write), [`writelines()`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamWriter.writelines), [`can_write_eof()`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamWriter.can_write_eof), [`write_eof()`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamWriter.write_eof), [`get_extra_info()`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamWriter.get_extra_info) и [`close()`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamWriter.close). Он добавляет [`drain()`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamWriter.drain), который возвращает необязательный [`Future`](https://python-all.ru/3.5/library/asyncio-task.html#asyncio.Future), на котором можно ожидать управления потоком. Также добавляется атрибут транспорта, который ссылается непосредственно на `Transport`.

This class is [не является потокобезопасным](https://python-all.ru/3.5/library/asyncio-dev.html#asyncio-multithreading).

#### `transport`

Транспорт.

#### `can_write_eof()`

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

#### `close()`

Закрывает транспорт: см. [`BaseTransport.close()`](https://python-all.ru/3.5/library/asyncio-protocol.html#asyncio.BaseTransport.close).

#### `coroutine drain()`

Даёт возможность сбросить буфер записи нижележащего транспорта.

Предполагаемое использование – записать:

```python
w.write(data)
yield from w.drain()
```

Когда размер буфера транспорта достигает верхнего порога (протокол приостанавливается), блокируется до тех пор, пока размер буфера не уменьшится до нижнего порога и протокол не будет возобновлён. Когда ждать нечего, yield-from продолжается немедленно.

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

Этот метод является [корутиной](https://python-all.ru/3.5/library/asyncio-task.html#coroutine).

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

Возвращает необязательную информацию о транспорте: см. [`BaseTransport.get_extra_info()`](https://python-all.ru/3.5/library/asyncio-protocol.html#asyncio.BaseTransport.get_extra_info).

#### `write(data)`

Записать некоторые *data* байтов в транспорт: см. [`WriteTransport.write()`](https://python-all.ru/3.5/library/asyncio-protocol.html#asyncio.WriteTransport.write).

#### `writelines(data)`

Записать список (или любой итерируемый объект) байтов в транспорт: см. [`WriteTransport.writelines()`](https://python-all.ru/3.5/library/asyncio-protocol.html#asyncio.WriteTransport.writelines).

#### `write_eof()`

Закрыть конец записи транспорта после сброса буферизованных данных: см. [`WriteTransport.write_eof()`](https://python-all.ru/3.5/library/asyncio-protocol.html#asyncio.WriteTransport.write_eof).

## 18.5.5.4. StreamReaderProtocol

#### `class asyncio.StreamReaderProtocol(stream_reader, client_connected_cb=None, loop=None)`

Вспомогательный класс для адаптации между [`Protocol`](https://python-all.ru/3.5/library/asyncio-protocol.html#asyncio.Protocol) и [`StreamReader`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamReader). Подкласс [`Protocol`](https://python-all.ru/3.5/library/asyncio-protocol.html#asyncio.Protocol).

*stream\_reader* является экземпляром [`StreamReader`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamReader), *client\_connected\_cb* – необязательная функция, вызываемая с (stream\_reader, stream\_writer) при установлении соединения, *loop* – используемый экземпляр цикла событий.

(Это вспомогательный класс вместо того, чтобы делать сам [`StreamReader`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamReader) подклассом [`Protocol`](https://python-all.ru/3.5/library/asyncio-protocol.html#asyncio.Protocol), так как [`StreamReader`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamReader) имеет другие потенциальные применения, и чтобы предотвратить случайный вызов пользователем [`StreamReader`](https://python-all.ru/3.5/library/asyncio-stream.html#asyncio.StreamReader) неподходящих методов протокола.)

## 18.5.5.5. IncompleteReadError

#### `exception asyncio.IncompleteReadError`

> Ошибка неполного чтения, подкласс
>
> [`EOFError`](https://python-all.ru/3.5/library/exceptions.html#EOFError)
>
> .

#### `expected`

Общее количество ожидаемых байтов ([`int`](https://python-all.ru/3.5/library/functions.html#int)).

#### `partial`

Прочитанная строка байтов до достижения конца потока ([`bytes`](https://python-all.ru/3.5/library/functions.html#bytes)).

## 18.5.5.6. LimitOverrunError

#### `exception asyncio.LimitOverrunError`

Достигнут предел буфера во время поиска разделителя.

#### `consumed`

Общее количество байтов для обработки.

## 18.5.5.7. Примеры потоков данных

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

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

```python
import asyncio

@asyncio.coroutine
def tcp_echo_client(message, loop):
    reader, writer = yield from asyncio.open_connection('127.0.0.1', 8888,
                                                        loop=loop)

    print('Send: %r' % message)
    writer.write(message.encode())

    data = yield from reader.read(100)
    print('Received: %r' % data.decode())

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

message = 'Hello World!'
loop = asyncio.get_event_loop()
loop.run_until_complete(tcp_echo_client(message, loop))
loop.close()
```

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

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

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

```python
import asyncio

@asyncio.coroutine
def handle_echo(reader, writer):
    data = yield from reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print("Received %r from %r" % (message, addr))

    print("Send: %r" % message)
    writer.write(data)
    yield from writer.drain()

    print("Close the client socket")
    writer.close()

loop = asyncio.get_event_loop()
coro = asyncio.start_server(handle_echo, '127.0.0.1', 8888, loop=loop)
server = loop.run_until_complete(coro)

# Обслуживать запросы до нажатия Ctrl+C
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Закрыть сервер
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()
```

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

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

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

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

@asyncio.coroutine
def print_http_headers(url):
    url = urllib.parse.urlsplit(url)
    if url.scheme == 'https':
        connect = asyncio.open_connection(url.hostname, 443, ssl=True)
    else:
        connect = asyncio.open_connection(url.hostname, 80)
    reader, writer = yield from connect
    query = ('HEAD {path} HTTP/1.0\r\n'
             'Host: {hostname}\r\n'
             '\r\n').format(path=url.path or '/', hostname=url.hostname)
    writer.write(query.encode('latin-1'))
    while True:
        line = yield from reader.readline()
        if not line:
            break
        line = line.decode('latin1').rstrip()
        if line:
            print('HTTP header> %s' % line)

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

url = sys.argv[1]
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(print_http_headers(url))
loop.run_until_complete(task)
loop.close()
```

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

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

или с HTTPS:

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

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

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

```python
import asyncio
try:
    from socket import socketpair
except ImportError:
    from asyncio.windows_utils import socketpair

@asyncio.coroutine
def wait_for_data(loop):
    # Создаём пару соединённых сокетов
    rsock, wsock = socketpair()

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

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

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

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

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

loop = asyncio.get_event_loop()
loop.run_until_complete(wait_for_data(loop))
loop.close()
```

> **См. также**
>
> Пример [регистрации открытого сокета для ожидания данных с помощью протокола](https://python-all.ru/3.5/library/asyncio-protocol.html#asyncio-register-socket) использует низкоуровневый протокол, созданный методом [`AbstractEventLoop.create_connection()`](https://python-all.ru/3.5/library/asyncio-eventloop.html#asyncio.AbstractEventLoop.create_connection).
>
> Пример [наблюдения за файловым дескриптором на предмет событий чтения](https://python-all.ru/3.5/library/asyncio-eventloop.html#asyncio-watch-read-event) использует низкоуровневый метод [`AbstractEventLoop.add_reader()`](https://python-all.ru/3.5/library/asyncio-eventloop.html#asyncio.AbstractEventLoop.add_reader) для регистрации файлового дескриптора сокета.