Расширение

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

Примечание

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

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

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

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

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

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

Частные конструкторы Future и Task

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

Однако сторонние циклы событий могут использовать встроенные реализации future и task ради того, чтобы получить сложный и высокооптимизированный код бесплатно.

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

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