Plus je fais joujou avec asyncio, plus j’apprécie la lib. Mais je tombe aussi sur des tas de petits trucs qui me font dire qu’il va falloir créer quelques couches d’abstraction pour rendre tout ça plus miam.
Par exemple, lire de manière asynchrone les données pipées sur stdin:
async def main(loop): reader = asyncio.StreamReader() def get_reader(): return asyncio.StreamReaderProtocol(reader) await loop.connect_read_pipe(get_reader, sys.stdin) while True: line = await reader.readline() if not line: break print(line) loop = asyncio.get_event_loop() loop.run_until_complete(main(loop)) |
C’est assez verbeux, pas très Pythonique (une factory qui instancie un classe a qui on passe une autre instance, c’est très java), ça n’exploite pas toute la force de async/await (le while qui pourrait être un async for), et il faut se soucier de lancer la boucle.
En prime, c’est pas encore évident de trouver qu’il faut faire ça dans la doc.
Pareil, pour lire de manière asynchrone les données écrites par un utilisateur sur stdin:
def on_stdin(*args): print("Somebody wrote:", sys.stdin.readline()) loop = asyncio.get_event_loop() loop.add_reader(sys.stdin.fileno(), on_stdin) loop.run_forever() |
Bon, là on a du callback, clairement on peut faire mieux.
La bonne nouvelle, c’est que ça veut dire qu’on a un champ entier où on peut être le premier à écrire une lib, et donc devenir une implémentation de référence, un outil connu, etc. Si vous avez envie de faire votre trou, c’est une opportunité, et en plus, c’est rigolo :)
En effet, avec un peu d’enrobage, on peut rapidement faire des trucs choupinets:
class AioStdinPipe: def __init__(self, loop=None): self.loop = loop or asyncio.get_event_loop() self.reader = asyncio.StreamReader() def get_reader(self): return asyncio.StreamReaderProtocol(self.reader) async def __aiter__(self): await self.loop.connect_read_pipe(self.get_reader, sys.stdin) return self async def __anext__(self): while True: val = await self.reader.readline() if val == b'': raise StopAsyncIteration return val def run_in_loop(coro): loop = asyncio.get_event_loop() loop.run_until_complete(coro()) |
Et ainsi réduire le premier code à un truc très simple, très clair, très Python:
@run_in_loop async def main(): async for line in AioStdinPipe(): print(line) |
J’aime 2016, c’est une année pleine de possibilités.
Salut Sam,
self.loop = loop or asyncio.get_event_loop()C’est une nouveauté de python 3 ça? Ça équivaut à un
if loop is not None: self.loop = loop
else: self.loop = asyncio.get_event_loop()
Non, ça a toujours marché comme ça : “or” retourne la première valeur vraie rencontrée ou la dernière valeur fausse. Tandis que “and” retourne la première valeur fausse rencontrée ou la dernière valeur vraie. C’est juste que quand on enseigne Python on explique pas ce genre de chose car les conditions c’est déjà un gros morceaux.
Merci pour ces quelques éclaircissements.
L’absence de tutoriel, la doc de références officielle brute de fonderie, ainsi que la culture “synchrone” des devs Python ne rendent pas la chose aisée à comprendre.
really nice!
C’est marrant j’ai écrit un truc similaire pour un projet annexe, c’est une coroutine qui rend des
asyncio.Stream:stdin_stream, stdout_stream = await get_standard_streams()