Ensure Python async generator and context manager closes with contextlib.aclosing

2024-09-18 (Wednesday) | 500 words (~3 minutes reading)

I noticed a ResourceWarning like this when using aiofiles to iterate lines of a file in an async context manager from aiofiles.open:

ResourceWarning: unclosed file
<_io.TextIOWrapper
	name='.../numbers.tmp'
	mode='r'
	encoding='UTF-8'>

This is only a warning, so by default it won’t error out of the Python program. I try to have pytest filterwarnings set to error whenever possible, so that any warnings during tests will cause the test to error out. That’s how I spotted this issue.

It’s caused by not completing an async generator, which then fails to exit the async context manager from aiofiles.open, which then leaves the file handle open until the end of the program.

Here’s a little Python script to reproduce the problem:

import os
from typing import AsyncGenerator

import aiofiles


# Write some lines to a file.
file_path = os.path.join(
    os.path.dirname(os.path.abspath(__file__)),
    "numbers.tmp",
)
async with aiofiles.open(file_path, "w") as wf:
    for i in range(10):
        await wf.write(str(i) + os.linesep)

# Use a generator to iterate all the lines in the file.
async def iter_file_lines() -> AsyncGenerator[str, None]:
    async with aiofiles.open(file_path, "r") as rf:
        async for rf_line in rf:
            yield rf_line

# This is fine, as it completes the async generator,
# which exits the async context manager, which closes
# the file handle.
async for line in iter_file_lines():
    print(line)

# Try to take only the first item from the generator.
async def get_first_line_only() -> str:
    async for iter_line in iter_file_lines():
        return iter_line

# This will cause a `ResourceWarning: unclosed file` warning.
first_line = await get_first_line_only()
print(first_line)

The easiest way to prevent this might be to use the aclosing utility from the contextlib library to wrap the async context manager. The aclosing wrapper ensures that the aexit method on the async context manager is always called, regardless of exceptions or unfinished generators.

Continuing from the above example, that looks like this:

# Take only the first item from the generator, and correctly close the
# async generator.
async def get_first_line_only() -> str:
    async with aclosing(iter_file_lines()) as iter_lines:
        async for iter_line in iter_lines:
            return iter_line

# This will not raise any warnings.
first_line = await get_first_line_only()
print(first_line)

Alternatively, you can take the next value from the async generator and then manually close the generator like this:

# Take only the first item from the generator,
# and correctly close the async generator.
async def get_first_line_only() -> str:
    lines_generator = iter_file_lines()
    next_item = await anext(lines_generator)
    await lines_generator.aclose()
    return next_item

# This will not raise any warnings.
first_line = await get_first_line_only()
print(first_line)

This feels a bit more verbose and procedural than using contextlib.aclosing.

Hire me as a freelance Python developer.


Tech mentioned