Skip to content

Appearance

Theme

Foxipa Forest

Mode

Light Dark System

Language selection

English Français العربية

Code Completion 5 Challenges

Complete the Python asynchronous generator code

Practice defining asynchronous generators, preserving state across yields, stopping early, and using aclose() or contextlib.aclosing() for deterministic cleanup.

Intermediate ~8 min read

Turn this function into an asynchronous generator that yields each value after a suspension point.

python Your answer

import asyncio

async def numbers(values):
    for value in values:
        

async def main():
    print([value async for value in numbers([1, 2, 3])])

asyncio.run(main())

Press Ctrl⌘Enter to check your answer

Preserve the running total between yields.

python Your answer

import asyncio

async def totals(values):
    total = 0
    for value in values:
        total += value
        

async def main():
    print([value async for value in totals([3, 4, 5])])

asyncio.run(main())

Press Ctrl⌘Enter to check your answer

Stop the asynchronous generator as soon as a value greater than 2 is reached.

python Your answer

async def first_small(values):
    for value in values:
        if value > 2:
            
        yield value

import asyncio

async def main():
    print([value async for value in first_small([1, 2, 3, 4])])

asyncio.run(main())

Press Ctrl⌘Enter to check your answer

Ensure cleanup runs when the generator is explicitly closed.

python Your answer

async def stream(events):
    try:
        yield "item"
    

import asyncio

async def main():
    events = []
    generator = stream(events)
    await generator.__anext__()
    await generator.aclose()
    print(events)

asyncio.run(main())

Press Ctrl⌘Enter to check your answer

Wrap the asynchronous generator in a context manager so its finally block runs immediately when the consumer breaks out of async for.

python Your answer

import asyncio
from contextlib import aclosing

async def source(events):
    try:
        yield "ready"
        yield "later"
    finally:
        await asyncio.sleep(0)
        events.append("closed")

async def main():
    events = []
    generator = source(events)
    async with  as items:
        async for item in items:
            print(item)
            break
    print(events)

asyncio.run(main())

Press Ctrl⌘Enter to check your answer