Потоки данных¶Streams

StreamReader¶

class asyncio.StreamReader¶

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

Не рекомендуется создавать объекты StreamReader напрямую; вместо этого используйте open_connection() и start_server().

coroutine read(n=-1)¶

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

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

coroutine readline()¶

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

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

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

coroutine readexactly(n)¶

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

Возбуждается IncompleteReadError, если EOF достигнут до того, как прочитано n байт. Используйте атрибут IncompleteReadError.partial, чтобы получить частично прочитанные данные.

coroutine readuntil(separator=b'\n')¶

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

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

Если объем прочитанных данных превышает настроенный лимит потока, возбуждается исключение LimitOverrunError, а данные остаются во внутреннем буфере и могут быть прочитаны снова.

Если EOF достигнут до того, как найден полный разделитель, возбуждается исключение IncompleteReadError, и внутренний буфер сбрасывается. Атрибут IncompleteReadError.partial может содержать часть разделителя.

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

at_eof()¶

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

StreamWriter¶

class asyncio.StreamWriter¶

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

Не рекомендуется создавать объекты StreamWriter напрямую; вместо этого используйте open_connection() и start_server().

can_write_eof()¶

Возвращает True, если нижележащий транспорт поддерживает метод write_eof(), и False в противном случае.

write_eof()¶

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

transport¶

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

get_extra_info(name, default=None)¶

Предоставляет доступ к дополнительной информации о транспорте; подробнее см. BaseTransport.get_extra_info().

write(data)¶

Записывает data в поток.

Этот метод не подчиняется управлению потоком. Вызовы write() должны сопровождаться вызовом drain().

writelines(data)¶

Записывает в поток список (или любой итерируемый объект) байтов.

Этот метод не подчиняется управлению потоком. Вызовы writelines() должны сопровождаться вызовом drain().

coroutine drain()¶

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

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

Это метод управления потоком, который взаимодействует с базовым буфером записи ввода-вывода. Когда размер буфера достигает верхней границы, drain() блокируется, пока размер буфера не уменьшится до нижней границы, после чего запись можно возобновить. Если ждать нечего, drain() возвращает управление немедленно.

close()¶

Закрывает поток.

is_closing()¶

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

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

coroutine wait_closed()¶

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

Должен вызываться после close(), чтобы дождаться закрытия нижележащего соединения.

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

Примеры¶Examples

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!'))

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)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

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 example.py http://example.com/path/page.html

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

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())