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

Исходный код: Lib/asyncio/streams.py

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

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

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

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

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

Дополнительные необязательные ключевые аргументы — 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 с обратным вызовом для каждого подключённого клиента.

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

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

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

18.5.5.2. StreamReader

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

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

Значение по умолчанию для аргумента limit установлено в _DEFAULT_LIMIT, что составляет 2**16 (64 КБ)

exception()

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

feed_eof()

Подтверждение EOF.

feed_data(data)

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

set_exception(exc)

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

set_transport(transport)

Установка транспорта.

coroutine read(n=-1)

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

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

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

coroutine readline()

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

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

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

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

coroutine readexactly(n)

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

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

coroutine readuntil(separator=b'\n')

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

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

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

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

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

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

at_eof()

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

18.5.5.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() предоставляет возможность циклу запланировать операцию записи и очистить буфер. Это особенно полезно, когда в транспорт записывается большое количество данных, и корутина не передаёт управление между вызовами write().

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

get_extra_info(name, default=None)

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

write(data)

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

writelines(data)

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

write_eof()

Закрыть пишущий конец транспорта после сброса данных из буфера: см. WriteTransport.write_eof().

18.5.5.4. StreamReaderProtocol

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 не случайно вызывал неподходящие методы протокола.)

18.5.5.5. ОшибкаНеполногоЧтения

exception asyncio.IncompleteReadError

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

expected

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

partial

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

18.5.5.6. ОшибкаПревышенияЛимита

exception asyncio.LimitOverrunError

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

consumed

Общее количество байтов, подлежащих обработке.

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

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

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

18.5.5.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

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