Async iterators and context managers in Python

The async twins of the two protocols you already know.

import asyncio
from contextlib import asynccontextmanager


class Ticker:
    def __init__(self, count):
        self.count = count

    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 session(name):
    print(f"open {name}")
    try:
        yield name
    finally:
        print(f"close {name}")


async def main():
    async with session("run") as name:
        async for tick in Ticker(3):
            print(name, tick)


asyncio.run(main())

How it works

1. __aiter__ and __anext__ power async for.
2. StopAsyncIteration ends the loop.
3. asynccontextmanager builds an async with from a generator.

Keywords and builtins used here

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

The run, in numbers

Lines

35

Characters to type

525

Tokens

137

Three-star pace

110 tpm

At the three-star pace of 110 tokens a minute, this run takes about 75 seconds.

Type this snippet

Step 2 of 4 in Async in depth, step 49 of 53 in Pythonic Python.

← Previous Next →