Потоки

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

Вот пример TCP-эхо-клиента, написанного с использованием потоков asyncio:

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

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

См. также раздел Примеры ниже.

Функции потоков

Следующие функции 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)

Установить сетевое подключение и вернуть пару (reader, writer) объектов.

Возвращаемые объекты reader и writer являются экземплярами классов StreamReader и StreamWriter.

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

limit определяет предел размера буфера, используемый экземпляром возвращаемого объекта StreamReader. По умолчанию limit установлен в 64 КБ.

Остальные аргументы передаются напрямую в loop.create_connection().

Введено в версии 3.7: Параметр ssl_handshake_timeout.

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 и StreamWriter.

client_connected_cb может быть обычной вызываемой функцией или функцией-корутиной; если это функция-корутина, она будет автоматически запланирована как Task.

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

limit определяет предел размера буфера, используемый экземпляром возвращаемого объекта StreamReader. По умолчанию limit установлен в 64 КБ.

Остальные аргументы передаются напрямую в 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(), но работает с сокетами Unix.

См. также документацию по loop.create_unix_connection().

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

Введено в версии 3.7: Параметр ssl_handshake_timeout.

Изменено в версии 3.7: Параметр path теперь может быть объектом, подобным пути

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(), но работает с сокетами Unix.

См. также документацию по loop.create_unix_server().

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

Введено в версии 3.7: Параметры ssl_handshake_timeout и start_serving.

Изменено в версии 3.7: Параметр path теперь может быть объектом, подобным пути.

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.

Примеры

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

TCP-клиент эха, использующий функцию asyncio.open_connection():

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-сервер эха с использованием потоков

TCP-сервер эха, использующий функцию asyncio.start_server():

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

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

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

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

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

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

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

или с HTTPS:

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

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

Подпрограмма, ожидающая данных от сокета с использованием функции open_connection():

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

    # Close the second socket
    wsock.close()

asyncio.run(wait_for_data())