Python async generators: close a suspended producer when leaving early

Operation contract

The review producer records a cleanup marker in finally. The consumer exits after its first accepted receipt and wraps the producer with aclosing. Leaving that scope awaits aclose and executes the producer’s cleanup before the consumer returns. The fixture observes the marker directly rather than relying on garbage collection timing.

Failure and ownership boundary

Breaking an async-for loop does not establish prompt resource release for every retained generator reference. The generator may also fail during cleanup; that failure needs an explicit application policy. Aclosing supplies deterministic closure, not cancellation shielding or a network deadline. Python asyncio timeout: cancellation and cleanup ownership and Python concurrency interview: cancellation must still run cleanup cover those boundaries.

Working program

import asyncio
from contextlib import aclosing

async def review_stream(released):
    try:
        yield "R-0041"
        yield "R-0042"
    finally:
        released.append("producer closed")

async def first_review():
    released = []
    async with aclosing(review_stream(released)) as stream:
        async for receipt_id in stream:
            print(receipt_id)
            break
    print(released)

asyncio.run(first_review())

Output

R-0041
['producer closed']

Costs and limits

Only one value is consumed in this fixture. Producer work, buffered state and awaited cleanup cost depend on the actual generator. Deterministic closure removes reliance on garbage collection but does not place a bound on arbitrary cleanup work.

Common Mistakes

• Early loop exit is not a resource-release policy by itself.
• Await closure before claiming the producer has released its resources.

Connected lessons

Python generators: lazy iteration does not make retained output free, Python asyncio timeout: cancellation and cleanup ownership, Python concurrency interview: cancellation must still run cleanup.