Потоки

StreamReader

class asyncio.StreamReader

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

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

feed_eof()

Сообщить о достижении EOF.

async read(n=-1)

Прочитать из потока не более n байт.

Если n не задано или равно -1, читать до EOF, а затем вернуть все прочитанные данные типа bytes. Если получен EOF и внутренний буфер пуст, вернуть пустой объект bytes.

Если n равно 0, немедленно вернуть пустой объект bytes.

Если n положительно, вернуть не более n доступных bytes, как только во внутреннем буфере окажется хотя бы 1 байт. Если EOF получен до чтения хотя бы одного байта, вернуть пустой объект bytes.

async readline()

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

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

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

async readexactly(n)

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

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

async readuntil(separator=b'\n')

Читать данные из потока до обнаружения separator.

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

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

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

separator также может быть кортежем разделителей. В этом случае возвращаемое значение будет самым коротким из возможных, заканчивающимся на любой из разделителей. Для целей исключения LimitOverrunError совпавшим считается самый короткий из возможных разделителей.

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

Изменено в версии 3.13: Теперь параметр separator может быть tuple разделителей.

at_eof()

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

StreamWriter

class asyncio.StreamWriter

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

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

write(data)

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

Буфер data должен быть объектом bytes, bytearray или одномерным объектом memoryview с непрерывным расположением в памяти в стиле C.

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

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

writelines(data)

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

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

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

close()

Метод закрывает поток и базовый сокет.

Метод рекомендуется, но не обязательно использовать вместе с методом wait_closed():

stream.close()
await stream.wait_closed()

can_write_eof()

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

write_eof()

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

transport

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

get_extra_info(name, default=None)

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

async drain()

Дождаться момента, когда можно будет продолжить запись в поток. Пример:

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

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

async start_tls(sslcontext, *, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None)

Перевести существующее соединение на основе потока на TLS.

Параметры:

• sslcontext: настроенный экземпляр SSLContext.
• server_hostname: задаёт или переопределяет имя узла, с которым будет сравниваться сертификат целевого сервера.
• ssl_handshake_timeout — время в секундах ожидания завершения TLS-рукопожатия до прерывания соединения. Если значение None (по умолчанию), используется 60.0 секунд.
• ssl_shutdown_timeout — время в секундах ожидания завершения отключения SSL до прерывания соединения. Если значение None (по умолчанию), используется 30.0 секунд.

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

Изменено в версии 3.12: Добавлен параметр ssl_shutdown_timeout.

is_closing()

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

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

async wait_closed()

Дождаться закрытия потока.

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

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

Примеры

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

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()
    await writer.wait_closed()

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

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

    # Ignore the body, close the socket
    writer.close()
    await writer.wait_closed()

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():
    # Get a reference to the current event loop because
    # we want to access low-level APIs.
    loop = asyncio.get_running_loop()

    # Create a pair of connected sockets.
    rsock, wsock = socket.socketpair()

    # Register the open socket to wait for data.
    reader, writer = await asyncio.open_connection(sock=rsock)

    # Simulate the reception of data from the network
    loop.call_soon(wsock.send, 'abc'.encode())

    # Wait for data
    data = await reader.read(100)

    # Got data, we are done: close the socket
    print("Received:", data.decode())
    writer.close()
    await writer.wait_closed()

    # Close the second socket
    wsock.close()

asyncio.run(wait_for_data())