Потоки (API на основе корутин)

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

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

Обёртка для create_connection(), возвращающая пару (чтец, писатель).

Возвращаемый чтец — это экземпляр StreamReader; писатель — это экземпляр StreamWriter.

Аргументы — все обычные аргументы для AbstractEventLoop.create_connection(), за исключением protocol_factory; наиболее распространены позиционные хост и порт, за которыми следуют различные необязательные ключевые аргументы.

Дополнительные необязательные ключевые аргументы — loop (для установки экземпляра цикла событий) и limit (для установки предела буфера, передаваемого в StreamReader).

Эта функция является корутиной.

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

Запуск сервера сокетов с обратным вызовом для каждого подключённого клиента. Возвращаемое значение такое же, как у create_server().

Параметр client_connected_cb вызывается с двумя параметрами: client_reader, client_writer. client_reader — это объект StreamReader, а client_writer — объект StreamWriter. Параметр client_connected_cb может быть обычной функцией обратного вызова или функцией-корутиной; если это функция-корутина, она будет автоматически преобразована в Task.

Остальные аргументы — все обычные аргументы для create_server(), за исключением protocol_factory; наиболее распространены позиционные host и port, за которыми следуют различные необязательные ключевые аргументы.

Дополнительные необязательные ключевые аргументы — loop (для установки экземпляра цикла событий) и limit (для установки предела буфера, передаваемого в StreamReader).

Эта функция является корутиной.

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

Обёртка для create_unix_connection(), возвращающая пару (чтец, писатель).

См. open_connection() для получения информации о возвращаемом значении и других деталях.

Эта функция является корутиной.

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

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

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

См. start_server() для получения информации о возвращаемом значении и других деталях.

Эта функция является корутиной.

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

2. StreamReader

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

Этот класс не потокобезопасен.

exception()

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

feed_eof()

Подтвердить конец потока.

feed_data(data)

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

set_exception(exc)

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

set_transport(transport)

Установить транспорт.

coroutine read(n=-1)

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

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

Этот метод является корутиной.

coroutine readline()

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

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

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

Этот метод является корутиной.

coroutine readexactly(n)

Прочитать ровно n байтов. Поднять IncompleteReadError, если конец потока достигнут до чтения n байтов, а атрибут IncompleteReadError.partial исключения содержит частично прочитанные байты.

Этот метод является корутиной.

coroutine readuntil(separator=b'\n')

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

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

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

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

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

New in version 3.5.2.

at_eof()

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

3. StreamWriter

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

Оборачивает транспорт.

Это предоставляет write(), writelines(), can_write_eof(), write_eof(), get_extra_info() и close(). Оно добавляет drain(), которое возвращает необязательное Future, на котором можно ожидать управления потоком. Также добавляется атрибут transport, который ссылается на Transport напрямую.

Этот класс не потокобезопасен.

transport

Транспорт.

can_write_eof()

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

close()

Закрыть транспорт: см. BaseTransport.close().

coroutine drain()

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

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

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

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

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

Этот метод является корутиной.

get_extra_info(name, default=None)

Возвращает необязательную информацию о транспорте: см. BaseTransport.get_extra_info().

write(data)

Записать некоторые байты данных в транспорт: см. WriteTransport.write().

writelines(data)

Записать список (или любой итерируемый объект) байтов данных в транспорт: см. WriteTransport.writelines().

write_eof()

Закрыть выходную часть транспорта после сброса данных из буфера: см. WriteTransport.write_eof().

4. Протокол StreamReader

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

Тривиальный вспомогательный класс для адаптации между Protocol и StreamReader. Подкласс Protocol.

stream_reader — экземпляр StreamReader, client_connected_cb — необязательная функция, вызываемая с (stream_reader, stream_writer) при подключении, loop — экземпляр цикла событий.

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

5. Ошибка IncompleteRead

exception asyncio.IncompleteReadError

Ошибка неполного чтения, подкласс EOFError.

expected

Общее количество ожидаемых байтов (int).

partial

Прочитанная строка байтов до достижения конца потока (bytes).

6. Ошибка LimitOverrun

exception asyncio.LimitOverrunError

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

consumed

Общее количество потребляемых байтов.

7. Примеры с потоками

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

Клиент TCP эха с использованием функции asyncio.open_connection():

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

7.2. Сервер TCP эха с использованием потоков

Сервер TCP эха с использованием функции asyncio.start_server():

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)

# Serve requests until Ctrl+C is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

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

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

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)

    # Ignore the body, close the socket
    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 example.py http://example.com/path/page.html

или с HTTPS:

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

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

Корутина ожидания, пока сокет не получит данные, используя функцию open_connection():

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

@asyncio.coroutine
def wait_for_data(loop):
    # Create a pair of connected sockets
    rsock, wsock = socketpair()

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

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

    # Wait for data
    data = yield from reader.read(100)

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

    # Close the second socket
    wsock.close()

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