Extensión#

La dirección principal para extender "asyncio" es escribir clases event loop personalizadas. Asyncio tiene ayudantes que podrían usarse para simplificar esta tarea.

Nota:

Los terceros deben reutilizar el código asyncio existente con precaución, una nueva versión de Python es gratuita para romper la compatibilidad con versiones anteriores en la parte internal de la API.

Escribir un bucle de eventos personalizado#

"asyncio.AbstractEventLoop" declara muchos métodos. Implementarlos todos desde cero es un trabajo tedioso.

Un bucle puede obtener la implementación de muchos métodos comunes de forma gratuita al heredar de "asyncio.BaseEventLoop".

A su vez, el sucesor debería implementar un montón de métodos privados declarados pero no implementados en "asyncio.BaseEventLoop".

Por ejemplo, "loop.create_connection()" comprueba los argumentos, resuelve las direcciones DNS y llama a "loop._make_socket_transport()" que debería implementar la clase heredada. El método "_make_socket_transport()" no está documentado y se considera como una API internal.

Constructores privados Future y Task#

"asyncio.Future" y "asyncio.Task" nunca deben crearse directamente; en su lugar, utilice las fábricas "loop.create_future()" y "loop.create_task()" o "asyncio.create_task()" correspondientes.

Sin embargo, event loops de terceros puede reuse incorporar futuras implementaciones y tareas con el fin de obtener un código complejo y altamente optimizado de forma gratuita.

Para ello se listan los siguientes constructores private:

Future.init(*, loop=None)

Cree una instancia futura integrada.

loop es una instancia de bucle de eventos opcional.

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

Cree una instancia de tarea integrada.

loop es una instancia de bucle de eventos opcional. El resto de argumentos se describen en la descripción de "loop.create_task()".

Distinto en la versión 3.11: Se agrega el argumento context.

Soporte de por vida de tareas#

La implementación de una tarea de terceros debe llamar a las siguientes funciones para mantener una tarea visible para "asyncio.all_tasks()" y "asyncio.current_task()":

asyncio._register_task(task)

Registre un nuevo task como administrado por asyncio.

Llame a la función desde un constructor de tareas.

asyncio._unregister_task(task)

Anule el registro de un task de las estructuras internas de asyncio.

La función debe llamarse cuando una tarea está a punto de finalizar.

asyncio._enter_task(loop, task)

Cambie la tarea actual al argumento task.

Llame a la función justo antes de ejecutar una parte del coroutine incrustado ("coroutine.send()" o "coroutine.throw()").

asyncio._leave_task(loop, task)

Vuelva a cambiar la tarea actual de task a "None".

Llame a la función justo después de la ejecución de "coroutine.send()" o "coroutine.throw()".