mrkeyoor.com_
Thu 06 Aug 05:56 UTC
PyPIUtilsupdated 06 Aug 2026

aiofiles

aiofiles gives asyncio code a file API that looks exactly like the builtin one, except the calls are coroutines. You write async with aiofiles.open(path) as f and then await f.read(), and the actual blocking work is handed to a thread pool executor so your event loop stays free to serve other tasks. It also ships aiofiles.os with awaitable versions of common os functions (stat, rename, listdir, scandir, makedirs, sendfile and friends), aiofiles.tempfile mirroring the tempfile module, and async wrappers around stdin, stdout, and stderr. There is no kernel-level async IO involved: this is a careful, well-tested thread offload with a familiar surface.

Verdict

The straightforward way to keep file IO off your event loop, and the API match with builtin open makes porting nearly free. Just be clear that it buys concurrency, not speed, and check whether anyio already in your stack makes it redundant.

API stability5/5The open() signature deliberately tracks the builtin and has not changed shape in years; the only recent breaks are Python version floors, with 3.8 dropped in 25.1.0 and 3.7 in 24.1.0
Docs3/5The README is clear and lists every wrapped method and os function, plus a worked example of mocking it in tests, but that README is the entire documentation: no docs site, no API reference, and nothing about executor sizing or when threads stop paying off
Maintenance4/5Pushed 2026-07-18 with a 25.2.0 entry already in the changelog, a maintained CHANGELOG, and Python 3.14 support added in 25.1.0; releases come about once a year and it is effectively a single-maintainer project with 50 open issues
Ecosystem4/5Around 38M downloads a week and a common transitive dependency of async web and cloud libraries; the async file space is fragmented though, with anyio, trio, and aiopath all covering overlapping ground

Use it if

  • You have an async web handler or worker that reads or writes files large enough that a blocking call would stall every other request on the loop
  • You are porting synchronous file code into asyncio and want a mechanical change, since the open signature, modes, iteration, and methods all match the builtin
  • You need async temporary files or directories, which aiofiles.tempfile provides for TemporaryFile, NamedTemporaryFile, SpooledTemporaryFile, and TemporaryDirectory
  • You want zero dependencies: the package is pure Python and pulls nothing else in
Skip it if

Setup reality

pip install aiofiles takes a second, pulls no dependencies, and needs Python 3.9 or newer (3.8 was dropped in 25.1.0, so pin 24.1.0 if you are stuck there). The gotchas are all in how you use it. Everything is a coroutine, so a forgotten await gives you a coroutine object instead of your file contents and no error until something else chokes on it. The open() call itself must be awaited or used with async with; a bare aiofiles.open(path) does nothing. Work runs on the loop's default executor unless you pass your own, which is the setting you actually want under load. And because the wheel has no py.typed marker, type checkers treat every call as untyped until you install types-aiofiles.

Patterns

Read a file without blocking the loopread-file

import aiofiles

async def load(path):
    async with aiofiles.open(path, mode='r') as f:
        return await f.read()

Both the async with and the await matter. aiofiles.open() returns an awaitable context manager, so calling it without either gives you an object that never touches the disk.

Stream a file line by lineiterate-lines

import aiofiles

async def count_errors(path):
    total = 0
    async with aiofiles.open(path) as f:
        async for line in f:
            if 'ERROR' in line:
                total += 1
    return total

async for keeps memory flat, but each line is still one executor round trip. For a million-line file, read in chunks instead and split in Python.

Write text and binary outputwrite-file

import aiofiles

async def save(path, payload: bytes):
    async with aiofiles.open(path, mode='wb') as f:
        await f.write(payload)
        await f.flush()

async def save_lines(path, lines):
    async with aiofiles.open(path, mode='w', encoding='utf-8') as f:
        await f.writelines(lines)

flush() is a coroutine too and must be awaited. Leaving the async with block closes the file, but if you need the bytes durable before the next step, flush explicitly first.

Copy a large file in chunkschunked-copy

import aiofiles

async def copy(src, dst, chunk_size=1 << 20):
    async with aiofiles.open(src, 'rb') as fin, aiofiles.open(dst, 'wb') as fout:
        while chunk := await fin.read(chunk_size):
            await fout.write(chunk)

Chunk size sets your overhead: 1 MB means one thread hop per megabyte, while 4 KB means hundreds of hops and worse throughput than a synchronous copy.

Use the async os functionsasync-os-calls

import aiofiles.os

async def prepare(path):
    await aiofiles.os.makedirs('/data/out', exist_ok=True)
    if await aiofiles.os.path.exists(path):
        info = await aiofiles.os.stat(path)
        if info.st_size == 0:
            await aiofiles.os.remove(path)
    return await aiofiles.os.listdir('/data/out')

import aiofiles alone does not give you aiofiles.os; import the submodule. The list of wrapped functions is fixed, and platform-specific ones like sendfile and statvfs only exist where the stdlib has them.

Async temporary files and directoriestemp-files

import os
import aiofiles.tempfile

async def stage(payload: bytes):
    async with aiofiles.tempfile.NamedTemporaryFile('wb+') as f:
        await f.write(payload)
        await f.seek(0)
        async for line in f:
            handle(line)

    async with aiofiles.tempfile.TemporaryDirectory() as d:
        target = os.path.join(d, 'file.ext')
        await process(target)

Everything the context manager yields is async, including seek and the iteration. The directory name it yields is a plain string, so join paths with the normal os.path helpers.

Give file IO its own thread poolcustom-executor

import asyncio
from concurrent.futures import ThreadPoolExecutor
import aiofiles

io_pool = ThreadPoolExecutor(max_workers=8, thread_name_prefix='fileio')

async def read_report(path):
    async with aiofiles.open(path, executor=io_pool) as f:
        return await f.read()

The default executor is shared with every other run_in_executor caller in the process and is capped near 32 threads. A dedicated pool keeps a burst of file work from starving unrelated blocking calls, and lets you size the disk concurrency deliberately.

Read many files at onceconcurrent-reads

import asyncio
import aiofiles

async def read_one(path, sem):
    async with sem:
        async with aiofiles.open(path) as f:
            return await f.read()

async def read_all(paths):
    sem = asyncio.Semaphore(16)
    return await asyncio.gather(*(read_one(p, sem) for p in paths))

Without the semaphore, gather over thousands of paths opens thousands of descriptors and queues thousands of executor jobs. Concurrency past the pool size buys nothing except memory pressure.

Read stdin and write stdout asynchronouslystdio-streams

import aiofiles

async def pipe_filter():
    async for line in aiofiles.stdin:
        if line.strip():
            await aiofiles.stdout.write(line.upper())
    await aiofiles.stdout.flush()

# binary variants: aiofiles.stdin_bytes, aiofiles.stdout_bytes, aiofiles.stderr_bytes

These wrap sys.stdin and friends in the same thread-offload machinery, so a slow producer on the other end of the pipe no longer freezes your loop. They are module-level objects, not something you open.

Mock file IO in testsmock-in-tests

from unittest import mock
import aiofiles
import aiofiles.threadpool

aiofiles.threadpool.wrap.register(mock.MagicMock)(
    lambda *args, **kwargs: aiofiles.threadpool.AsyncBufferedIOBase(*args, **kwargs)
)

async def test_writes_payload():
    mock_file = mock.MagicMock(read=lambda *a, **kw: b'chunk')
    with mock.patch('aiofiles.threadpool.sync_open', return_value=mock_file):
        async with aiofiles.open('filename', 'w') as f:
            await f.write('data')
        mock_file.write.assert_called_once_with('data')

Patching aiofiles.threadpool.sync_open is the documented seam, but the singledispatch registration is required as well or the wrapper does not know how to adapt a MagicMock. Registering it once at module import leaks into other tests in the same process.

Make type checkers understand ittyping-stubs

$ pip install types-aiofiles

# then this checks properly:
import aiofiles
from aiofiles.threadpool.text import AsyncTextIOWrapper

async def read(path: str) -> str:
    f: AsyncTextIOWrapper
    async with aiofiles.open(path) as f:
        return await f.read()

The wheel has no py.typed marker, so without the stubs mypy reports a missing-import error and pyright infers Any for every file object. Add types-aiofiles to your dev requirements, not your runtime ones.

Alternatives

PackageRegistryPick it when
anyioPyPIYou want the same thread-offloaded file API plus an async Path object, and code that runs unchanged on asyncio and trio
trioPyPIYou are already on trio, where trio.open_file and trio.Path cover this without a third-party package
aiopathPyPIYou want an async pathlib.Path replacement rather than an async open(), for code that mostly manipulates paths and metadata