Iteradores y administradores de contexto async en Python

Los gemelos async de los dos protocolos que ya conoces.

import asyncio
from contextlib import asynccontextmanager


class Contador:
    def __init__(self, conteo):
        self.count = conteo

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.count == 0:
            raise StopAsyncIteration
        self.count -= 1
        await asyncio.sleep(0)
        return self.count


@asynccontextmanager
async def sesion(nombre):
    print(f"abre {nombre}")
    try:
        yield nombre
    finally:
        print(f"cierra {nombre}")


async def principal():
    async with sesion("intento") as nombre:
        async for tic in Contador(3):
            print(nombre, tic)


asyncio.run(principal())

Cómo funciona

1. __aiter__ y __anext__ impulsan async for.
2. StopAsyncIteration termina el bucle.
3. asynccontextmanager arma un async with a partir de un generador.

Palabras clave y builtins usados aquí

• as
• async
• await
• class
• def
• finally
• for
• if
• print
• raise
• return
• self
• try
• with
• yield

El intento, en números

Líneas

35

Caracteres a escribir

554

Tokens

137

Ritmo de tres estrellas

110 tpm

Al ritmo de tres estrellas de 110 tokens por minuto, este intento toma unos 75 segundos.

Escribe este fragmento

Paso 2 de 4 en Async a fondo; paso 49 de 53 en Python pythónico.

← Anterior Siguiente →