Sam et Max

Se faciliter la vie quand on utilise asyncio dans le shell 2

lundi 7 mars 2016 à 15:03

Dernièrement j’ai mis les fonctions suivantes dans mon script PYTHONSTARTUP…

Toujours choper une event loop toute fraiche, et ouverte:

import asyncio
 
def aloop(*args, **kargs):
    """ Ensure they is an opened event loop available and return it"""
    loop = asyncio.get_event_loop()
    if loop.is_closed():
        policy = asyncio.get_event_loop_policy()
        loop = policy.new_event_loop(*args, **kargs)
        policy.set_event_loop(loop)
    return loop

Lancer une coroutine dans une event loop jusqu’à ce qu’elle se termine:

import inspect
 
def arun(coro):
    """ Run this in a event loop """
    loop = aloop()
    if not inspect.isawaitable(coro):
        if not inspect.iscoroutinefunction(coro):
            coro = asyncio.coroutine(coro)
        coro = coro()
    future = asyncio.ensure_future(coro)
    return loop.run_until_complete(future)

Et ça s’utilise comme ça:

async def foo():
    await asyncio.sleep(1)
    print('bar')
 
arun(coro)

Ca me fait gagner pas mal de temps.

J’utilise aussi cette commande magique dans iPython qui permet de balancer des await en plein milieu du shell. Je suis en train de voir pour ajouter un truc similaire à ptpyhon.

Je me suis fait aussi un petit script pour lancer vite fait n’importe quel script dans une boucle asyncio de manière transparente, mais j’ai pas le temps de poster ça aujourd’hui donc ça sera pour la prochaine fois.