Транспорты и протоколы¶Transports and Protocols

Иерархия транспортов¶Transports Hierarchy

class asyncio.BaseTransport¶

Базовый класс для всех транспортов. Содержит методы, общие для всех транспортов asyncio.

class asyncio.WriteTransport(BaseTransport)¶

Базовый транспорт для соединений только на запись.

Экземпляры класса WriteTransport возвращаются из метода цикла событий loop.connect_write_pipe(), а также используются методами, связанными с подпроцессами, такими как loop.subprocess_exec().

class asyncio.ReadTransport(BaseTransport)¶

Базовый транспорт для соединений только на чтение.

Экземпляры класса ReadTransport возвращаются из метода цикла событий loop.connect_read_pipe(), а также используются методами, связанными с подпроцессами, такими как loop.subprocess_exec().

class asyncio.Transport(WriteTransport, ReadTransport)¶

Интерфейс, представляющий двунаправленный транспорт, такой как TCP-соединение.

Пользователь не создаёт транспорт напрямую; он вызывает вспомогательную функцию, передавая фабрику протокола и другие сведения, необходимые для создания транспорта и протокола.

Экземпляры класса транспорт возвращаются или используются методами цикла событий, например loop.create_connection(), loop.create_unix_connection(), loop.create_server(), loop.sendfile() и т.д.

class asyncio.DatagramTransport(BaseTransport)¶

Транспорт для дейтаграммных (UDP) соединений.

Экземпляры класса DatagramTransport возвращаются из метода цикла событий loop.create_datagram_endpoint().

class asyncio.SubprocessTransport(BaseTransport)¶

Абстракция, представляющая соединение между родительским и дочерним процессом ОС.

Экземпляры класса SubprocessTransport возвращаются из методов цикла событий loop.subprocess_shell() и loop.subprocess_exec().

Базовый транспорт¶Base Transport

BaseTransport.close()¶

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

Если у транспорта есть буфер для исходящих данных, буферизованные данные будут сброшены асинхронно. Новые данные приниматься не будут. После того как все буферизованные данные будут сброшены, метод protocol.connection_lost() протокола будет вызван с None в качестве аргумента.

BaseTransport.is_closing()¶

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

BaseTransport.get_extra_info(name, default=None)¶

Возвращает информацию о транспорте или используемых им базовых ресурсах.

name – строка, представляющая запрашиваемую часть информации, специфичной для транспорта.

default – значение, возвращаемое, если информация недоступна, либо если транспорт не поддерживает её запрос с данной сторонней реализацией цикла событий или на текущей платформе.

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

sock = transport.get_extra_info('socket')
if sock is not None:
    print(sock.getsockopt(...))

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

• сокет:

  • 'peername': удалённый адрес, к которому подключён сокет, результат socket.socket.getpeername() (None в случае ошибки)

  • 'socket': экземпляр socket.socket

  • 'sockname': собственный адрес сокета, результат socket.socket.getsockname()

• SSL-сокет:

  • 'compression': используемый алгоритм сжатия в виде строки, или None, если соединение не сжато; результат ssl.SSLSocket.compression()

  • 'cipher': кортеж из трёх значений, содержащий название используемого шифра, версию протокола SSL, определяющую его использование, и количество используемых секретных бит; результат ssl.SSLSocket.cipher()

  • 'peercert': сертификат однорангового узла; результат ssl.SSLSocket.getpeercert()

  • 'sslcontext': экземпляр ssl.SSLContext

  • 'ssl_object': экземпляр ssl.SSLObject или ssl.SSLSocket

• канал:

  • 'pipe': объект канала

• подпроцесс:

  • 'subprocess': экземпляр subprocess.Popen

BaseTransport.set_protocol(protocol)¶

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

Переключать протокол следует только при условии, что оба протокола документированы как поддерживающие переключение.

BaseTransport.get_protocol()¶

Возвращает текущий протокол.

Транспорты только для чтения¶Read-only Transports

ReadTransport.is_reading()¶

Возвращает True, если транспорт получает новые данные.

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

ReadTransport.pause_reading()¶

Приостанавливает приёмную сторону транспорта. Данные не будут передаваться методу protocol.data_received() протокола до тех пор, пока не будет вызван resume_reading().

Изменено в версии 3.7: Метод идемпотентен, то есть его можно вызывать, когда транспорт уже приостановлен или закрыт.

ReadTransport.resume_reading()¶

Возобновляет приёмную сторону. Метод протокола protocol.data_received() будет вызван снова, если для чтения доступны какие-либо данные.

Изменено в версии 3.7: Метод идемпотентен, то есть его можно вызывать, когда транспорт уже читает.

Транспорты только для записи¶Write-only Transports

WriteTransport.abort()¶

Закрывает транспорт немедленно, не дожидаясь ожидающих операций до завершения. Буферизованные данные будут потеряны. Новые данные приниматься не будут. Метод protocol.connection_lost() протокола в конечном итоге будет вызван с аргументом None.

WriteTransport.can_write_eof()¶

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

WriteTransport.get_write_buffer_size()¶

Возвращает текущий размер выходного буфера, используемого транспортом.

WriteTransport.get_write_buffer_limits()¶

Get the high and low watermarks for write flow control. Return a tuple (low, high) where low and high are positive number of bytes.

Используйте set_write_buffer_limits() для установки границ.

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

WriteTransport.set_write_buffer_limits(high=None, low=None)¶

Устанавливает верхнюю и нижнюю границы для управления потоком записи.

Эти два значения (измеряются в количестве байтов) управляют, когда методы протокола protocol.pause_writing() и protocol.resume_writing() вызываются. Если заданы, нижняя граница должна быть меньше или равна верхней. Ни верхняя, ни нижняя не могут быть отрицательными.

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

Значения по умолчанию зависят от реализации. Если задан только верхний порог, нижний порог по умолчанию принимает зависящее от реализации значение, меньшее или равное верхнему. Установка верхнего порога в ноль приводит к тому, что нижний также становится нулевым, и pause_writing() будет вызываться всякий раз, когда буфер становится непустым. Установка нижнего порога в ноль приводит к тому, что resume_writing() будет вызываться только когда буфер опустеет. Использование нуля для любого порога в целом неоптимально, так как сокращает возможности для одновременного выполнения ввода-вывода и вычислений.

get_write_buffer_limits() используется для получения ограничений.

WriteTransport.write(data)¶

Записывает несколько байт данных в транспорт.

Этот метод не блокирует выполнение; он буферизует данные и обеспечивает их асинхронную отправку.

WriteTransport.writelines(list_of_data)¶

Записывает в транспорт список (или любой итерируемый объект) из байт данных. Функционально это эквивалентно вызову write() для каждого элемента, полученного из итерируемого объекта, но может быть реализовано более эффективно.

WriteTransport.write_eof()¶

Закрывает записывающий конец транспорта после сброса всех буферизованных данных. Данные всё ещё могут приниматься.

Этот метод может вызывать NotImplementedError, если транспорт (например, SSL) не поддерживает полузакрытые соединения.

Дейтаграммные транспорты¶Datagram Transports

DatagramTransport.sendto(data, addr=None)¶

Отправляет байты данных удаленному узлу, задаваемому addr (адрес назначения, зависящий от транспорта). Если addr равно None, данные отправляются на адрес назначения, указанный при создании транспорта.

Этот метод не блокирует выполнение; он буферизует данные и организует их асинхронную отправку.

DatagramTransport.abort()¶

Немедленно закрывает транспорт, не дожидаясь завершения ожидающих операций. Буферизованные данные будут потеряны. Приём данных прекратится. Метод protocol.connection_lost() протокола в конечном итоге будет вызван с None в качестве аргумента.

Транспорты подпроцессов¶Subprocess Transports

SubprocessTransport.get_pid()¶

Возвращает идентификатор процесса подпроцесса в виде целого числа.

SubprocessTransport.get_pipe_transport(fd)¶

Возвращает транспорт для канала связи, соответствующего целочисленному файловому дескриптору fd:

• 0: читаемый потоковый транспорт стандартного ввода (stdin) или None, если подпроцесс был создан без stdin=PIPE

• 1: записываемый потоковый транспорт стандартного вывода (stdout) или None, если подпроцесс был создан без stdout=PIPE

• 2: записываемый потоковый транспорт стандартного вывода ошибок (stderr) или None, если подпроцесс был создан без stderr=PIPE

• другой fd: None

SubprocessTransport.get_returncode()¶

Возвращает код возврата подпроцесса в виде целого числа или None, если подпроцесс ещё не завершился, что аналогично атрибуту subprocess.Popen.returncode.

SubprocessTransport.kill()¶

Завершает подпроцесс.

В системах POSIX функция отправляет подпроцессу сигнал SIGKILL. В Windows этот метод является псевдонимом для terminate().

См. также subprocess.Popen.kill().

SubprocessTransport.send_signal(signal)¶

Отправляет подпроцессу номер signal, как в subprocess.Popen.send_signal().

SubprocessTransport.terminate()¶

Останавливает подпроцесс.

На системах POSIX этот метод отправляет сигнал SIGTERM подпроцессу. В Windows вызывается функция Windows API TerminateProcess() для остановки подпроцесса.

См. также subprocess.Popen.terminate().

SubprocessTransport.close()¶

Завершает подпроцесс вызовом метода kill().

Если подпроцесс ещё не завершён, закрывает транспорты каналов stdin, stdout и stderr.

Базовые протоколы¶Base Protocols

class asyncio.BaseProtocol¶

Базовый протокол, содержащий методы, общие для всех протоколов.

class asyncio.Protocol(BaseProtocol)¶

Базовый класс для реализации потоковых протоколов (TCP, сокеты Unix и т.д.).

class asyncio.BufferedProtocol(BaseProtocol)¶

Базовый класс для реализации потоковых протоколов с ручным управлением буфером приёма.

class asyncio.DatagramProtocol(BaseProtocol)¶

Базовый класс для реализации дейтаграммных протоколов (UDP).

class asyncio.SubprocessProtocol(BaseProtocol)¶

Базовый класс для реализации протоколов, взаимодействующих с дочерними процессами (однонаправленные каналы).

Базовый протокол¶Base Protocol

Все протоколы asyncio могут реализовывать колбэки базового протокола.

Колбэки соединения

Колбэки соединения вызываются для всех протоколов ровно один раз при каждом успешном соединении. Все остальные колбэки протокола могут вызываться только между этими двумя методами.

BaseProtocol.connection_made(transport)¶

Вызывается при установлении соединения.

Аргумент transport – это транспорт, представляющий соединение. Протокол отвечает за хранение ссылки на свой транспорт.

BaseProtocol.connection_lost(exc)¶

Вызывается при потере или закрытии соединения.

The argument is either an exception object or None. The latter means a regular EOF is received, or the connection was aborted or closed by this side of the connection.

Колбэки управления потоком

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

Подробнее см. в документации метода set_write_buffer_limits() .

BaseProtocol.pause_writing()¶

Вызывается, когда буфер транспорта превышает верхнюю границу.

BaseProtocol.resume_writing()¶

Вызывается, когда буфер транспорта опускается ниже нижней границы.

Если размер буфера равен верхней границе, pause_writing() не вызывается: размер буфера должен быть строго больше.

И наоборот, resume_writing() вызывается, когда размер буфера равен нижней границе или меньше её. Эти граничные условия важны, чтобы всё работало ожидаемым образом, когда любая из границ равна нулю.

Потоковые протоколы¶Streaming Protocols

Методы событий, такие как loop.create_server(), loop.create_unix_server(), loop.create_connection(), loop.create_unix_connection(), loop.connect_accepted_socket(), loop.connect_read_pipe() и loop.connect_write_pipe(), принимают фабрики, возвращающие потоковые протоколы.

Protocol.data_received(data)¶

Вызывается при получении данных. data – это непустой bytes объект, содержащий входящие данные.

То, буферизуются ли данные, разбиваются на части или собираются заново, зависит от tранспорта. В общем случае не следует полагаться на конкретную семантику, а вместо этого делать парсинг общим и гибким. Однако данные всегда принимаются в правильном порядке.

Метод может вызываться произвольное количество раз, пока соединение открыто.

Однако protocol.eof_received() вызывается не более одного раза. После вызова eof_received(), data_received() больше не вызывается.

Protocol.eof_received()¶

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

Этот метод может вернуть ложное значение (включая None), и в этом случае транспорт закроется сам. И наоборот, если метод возвращает истинное значение, используемый протокол определяет, закрывать ли транспорт. Поскольку реализация по умолчанию возвращает None, она неявно закрывает соединение.

Некоторые транспорты, включая SSL, не поддерживают полузакрытые соединения, и в этом случае возврат истинного значения из этого метода приведёт к закрытию соединения.

Автомат состояний:

start -> connection_made
    [-> data_received]*
    [-> eof_received]?
-> connection_lost -> end

Буферизованные потоковые протоколы¶Buffered Streaming Protocols

Буферизованные протоколы могут использоваться с любым методом цикла событий, который поддерживает потоковые протоколы.

Реализации BufferedProtocol позволяют явное ручное выделение и управление буфером приёма. Циклы событий могут затем использовать буфер, предоставленный протоколом, чтобы избежать лишнего копирования данных. Это может привести к заметному повышению производительности для протоколов, которые получают большие объёмы данных. Сложные реализации протоколов могут значительно сократить количество выделений буфера.

Следующие колбэки вызываются на экземплярах BufferedProtocol :

BufferedProtocol.get_buffer(sizehint)¶

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

sizehint – это рекомендуемый минимальный размер для возвращаемого буфера. Допускается возвращать буферы меньше или больше, чем предлагает sizehint. Если установлено -1, размер буфера может быть произвольным. Возвращать буфер нулевого размера – ошибка.

get_buffer() должен возвращать объект, реализующий буферный протокол.

BufferedProtocol.buffer_updated(nbytes)¶

Вызывается, когда буфер был обновлён полученными данными.

nbytes – общее количество байтов, записанных в буфер.

BufferedProtocol.eof_received()¶

См. документацию метода protocol.eof_received().

get_buffer() может вызываться произвольное число раз во время соединения. Однако protocol.eof_received() вызывается не более одного раза, и, если он вызван, get_buffer() и buffer_updated() не будут вызваны после него.

Автомат состояний:

start -> connection_made
    [-> get_buffer
        [-> buffer_updated]?
    ]*
    [-> eof_received]?
-> connection_lost -> end

Протоколы дейтаграмм¶Datagram Protocols

Экземпляры протокола дейтаграмм должны создаваться фабриками протоколов, передаваемыми методу loop.create_datagram_endpoint().

DatagramProtocol.datagram_received(data, addr)¶

Вызывается при получении дейтаграммы. data – это байтовый объект, содержащий входящие данные. addr – адрес узла, отправляющего данные; точный формат зависит от транспорта.

DatagramProtocol.error_received(exc)¶

Вызывается, когда предыдущая операция отправки или получения вызывает исключение OSError. exc – экземпляр OSError.

Этот метод вызывается в редких случаях, когда транспорт (например, UDP) обнаруживает, что дейтаграмма не может быть доставлена получателю. Однако во многих ситуациях недоставленные дейтаграммы молча отбрасываются.

Протоколы подпроцессов¶Subprocess Protocols

Экземпляры Datagram Protocol должны создаваться фабриками протоколов, переданными в методы loop.subprocess_exec() и loop.subprocess_shell().

SubprocessProtocol.pipe_data_received(fd, data)¶

Вызывается, когда дочерний процесс записывает данные в свой канал stdout или stderr.

fd – целочисленный файловый дескриптор канала.

data – непустой байтовый объект, содержащий полученные данные.

SubprocessProtocol.pipe_connection_lost(fd, exc)¶

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

fd – целочисленный файловый дескриптор, который был закрыт.

SubprocessProtocol.process_exited()¶

Вызывается, когда дочерний процесс завершился.

TCP-эхо-сервер¶TCP Echo Server

Создайте TCP-эхо-сервер с помощью метода loop.create_server(), отправьте обратно полученные данные и закройте соединение:

import asyncio


class EchoServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        message = data.decode()
        print('Data received: {!r}'.format(message))

        print('Send: {!r}'.format(message))
        self.transport.write(data)

        print('Close the client socket')
        self.transport.close()


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

    server = await loop.create_server(
        lambda: EchoServerProtocol(),
        '127.0.0.1', 8888)

    async with server:
        await server.serve_forever()


asyncio.run(main())

TCP-эхо-клиент¶TCP Echo Client

TCP-эхо-клиент, использующий метод loop.create_connection(), отправляет данные и ждёт, пока соединение не будет закрыто:

import asyncio


class EchoClientProtocol(asyncio.Protocol):
    def __init__(self, message, on_con_lost):
        self.message = message
        self.on_con_lost = on_con_lost

    def connection_made(self, transport):
        transport.write(self.message.encode())
        print('Data sent: {!r}'.format(self.message))

    def data_received(self, data):
        print('Data received: {!r}'.format(data.decode()))

    def connection_lost(self, exc):
        print('The server closed the connection')
        self.on_con_lost.set_result(True)


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

    on_con_lost = loop.create_future()
    message = 'Hello World!'

    transport, protocol = await loop.create_connection(
        lambda: EchoClientProtocol(message, on_con_lost),
        '127.0.0.1', 8888)

    # Ожидаем, пока протокол не сообщит, что соединение
    # потеряно, и закрываем транспорт.
    try:
        await on_con_lost
    finally:
        transport.close()


asyncio.run(main())

UDP-эхо-сервер¶UDP Echo Server

UDP-эхо-сервер, использующий метод loop.create_datagram_endpoint(), отправляет обратно полученные данные:

import asyncio


class EchoServerProtocol:
    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        message = data.decode()
        print('Received %r from %s' % (message, addr))
        print('Send %r to %s' % (message, addr))
        self.transport.sendto(data, addr)


async def main():
    print("Starting UDP server")

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

    # Будет создан один экземпляр протокола для обслуживания всех
    # запросов клиентов.
    transport, protocol = await loop.create_datagram_endpoint(
        lambda: EchoServerProtocol(),
        local_addr=('127.0.0.1', 9999))

    try:
        await asyncio.sleep(3600)  # Обслуживать в течение 1 часа.
    finally:
        transport.close()


asyncio.run(main())

UDP-эхо-клиент¶UDP Echo Client

UDP-эхо-клиент с помощью метода loop.create_datagram_endpoint() отправляет данные и закрывает транспорт при получении ответа:

import asyncio


class EchoClientProtocol:
    def __init__(self, message, on_con_lost):
        self.message = message
        self.on_con_lost = on_con_lost
        self.transport = None

    def connection_made(self, transport):
        self.transport = transport
        print('Send:', self.message)
        self.transport.sendto(self.message.encode())

    def datagram_received(self, data, addr):
        print("Received:", data.decode())

        print("Close the socket")
        self.transport.close()

    def error_received(self, exc):
        print('Error received:', exc)

    def connection_lost(self, exc):
        print("Connection closed")
        self.on_con_lost.set_result(True)


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

    on_con_lost = loop.create_future()
    message = "Hello World!"

    transport, protocol = await loop.create_datagram_endpoint(
        lambda: EchoClientProtocol(message, on_con_lost),
        remote_addr=('127.0.0.1', 9999))

    try:
        await on_con_lost
    finally:
        transport.close()


asyncio.run(main())

Подключение существующих сокетов¶Connecting Existing Sockets

Ожидание получения данных сокетом с помощью метода loop.create_connection() и протокола:

import asyncio
import socket


class MyProtocol(asyncio.Protocol):

    def __init__(self, on_con_lost):
        self.transport = None
        self.on_con_lost = on_con_lost

    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        print("Received:", data.decode())

        # Мы закончили: закрываем транспорт;
        # connection_lost() будет вызван автоматически.
        self.transport.close()

    def connection_lost(self, exc):
        # Сокет был закрыт
        self.on_con_lost.set_result(True)


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

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

    # Регистрируем сокет для ожидания данных.
    transport, protocol = await loop.create_connection(
        lambda: MyProtocol(on_con_lost), sock=rsock)

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

    try:
        await protocol.on_con_lost
    finally:
        transport.close()
        wsock.close()

asyncio.run(main())

loop.subprocess_exec() и SubprocessProtocol¶loop.subprocess_exec() and SubprocessProtocol

Пример протокола подпроцесса, используемого для получения вывода подпроцесса и ожидания его завершения.

Подпроцесс создаётся методом loop.subprocess_exec():

import asyncio
import sys

class DateProtocol(asyncio.SubprocessProtocol):
    def __init__(self, exit_future):
        self.exit_future = exit_future
        self.output = bytearray()

    def pipe_data_received(self, fd, data):
        self.output.extend(data)

    def process_exited(self):
        self.exit_future.set_result(True)

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

    code = 'import datetime; print(datetime.datetime.now())'
    exit_future = asyncio.Future(loop=loop)

    # Создаём подпроцесс, управляемый DateProtocol;
    # перенаправляем стандартный вывод в канал.
    transport, protocol = await loop.subprocess_exec(
        lambda: DateProtocol(exit_future),
        sys.executable, '-c', code,
        stdin=None, stderr=None)

    # Ожидаем завершения подпроцесса с помощью метода process_exited()
    # протокола.
    await exit_future

    # Закрываем канал stdout.
    transport.close()

    # Читаем вывод, который был собран методом
    # pipe_data_received() протокола.
    data = bytes(protocol.output)
    return data.decode('ascii').rstrip()

if sys.platform == "win32":
    asyncio.set_event_loop_policy(
        asyncio.WindowsProactorEventLoopPolicy())

date = asyncio.run(get_date())
print(f"Current date: {date}")

См. также тот же пример , написанный с использованием высокоуровневых API.