Расширяющий

Основным направлением расширения asyncio является написание пользовательских классов цикла обработки событий. В Asyncio есть помощники, которые могут быть использованы для упрощения этой задачи.

Примечание

Сторонним разработчикам следует с осторожностью использовать существующий код asyncio повторно, новая версия Python может нарушить обратную совместимость во внутренней части API.

Написание пользовательского цикла обработки событий

asyncio.AbstractEventLoop объявляет очень много методов. Реализация всех их с нуля - утомительная работа.

Цикл может получить реализацию многих распространенных методов бесплатно, унаследовав от asyncio.BaseEventLoop.

В свою очередь, преемник должен реализовать набор частных методов, объявленных, но не реализованных в asyncio.BaseEventLoop.

Например, loop.create_connection() проверяет аргументы, разрешает DNS-адреса и вызывает loop._make_socket_transport(), которые должны быть реализованы унаследованным классом. Метод _make_socket_transport() не документирован и рассматривается как внутренний API.

Будущее и задачи частных конструкторов

asyncio.Future и asyncio.Task никогда не должны создаваться напрямую, пожалуйста, используйте вместо них соответствующие loop.create_future() и loop.create_task() или asyncio.create_task() фабрики.

Однако сторонние event-циклы могут * повторно использовать* встроенные реализации функций и задач для бесплатного получения сложного и высокооптимизированного кода.

Для этой цели перечислены следующие частные конструкторы:

Future.__init__(*, loop=None)

Создайте встроенный будущий экземпляр.

loop - это необязательный экземпляр цикла обработки событий.

Task.__init__(coro, *, loop=None, name=None, context=None)

Создайте встроенный экземпляр задачи.

loop - это необязательный экземпляр цикла обработки событий. Остальные аргументы описаны в описании loop.create_task().

Изменено в версии 3.11: добавлен аргумент context.

Пожизненная поддержка задач

Сторонняя реализация задачи должна вызывать следующие функции, чтобы задача оставалась видимой с помощью asyncio.all_tasks() и asyncio.current_task():

asyncio._register_task(task)

Зарегистрируйте новую задачу под управлением asyncio.

Вызовите функцию из конструктора задачи.

asyncio._unregister_task(task)

Отмените регистрацию задачи из внутренних структур asyncio.

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

asyncio._enter_task(loop, task)

Переключите текущую задачу на аргумент task.

Вызовите функцию непосредственно перед выполнением части встроенной сопрограммы (coroutine.send() или coroutine.throw()).

asyncio._leave_task(loop, task)

Переключите текущую задачу обратно с task на None.

Вызовите функцию сразу после выполнения coroutine.send() или coroutine.throw().

Поддержка платформы
Высокоуровневый индекс API
Вернуться на верх