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.
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.
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
- You expected faster IO: every call is a round trip to a thread pool, so for small local files a plain open() finishes sooner than the scheduling overhead this adds
- You run on trio or want portability across async runtimes: aiofiles is asyncio-only, while anyio.open_file and anyio.Path do the same thread offload and work on both
- Your workload is bursty: the default executor is shared and bounded (CPython caps it around 32 threads), so a flood of file operations queues up behind it and delays every other run_in_executor caller in the process
- You want async pathlib: aiofiles.os exposes a fixed list of os functions with no Path object, so path-heavy code either stays synchronous or you write wrappers by hand
- You are serving static files from a web framework: FileResponse and equivalents already stream with os.sendfile, and routing that through aiofiles just adds thread hops
- You type-check strictly: no py.typed marker ships in the wheel, so mypy and pyright see an untyped module until you add the separate types-aiofiles stubs
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 totalasync 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_bytesThese 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
| Package | Registry | Pick it when |
|---|---|---|
| anyio | PyPI | You want the same thread-offloaded file API plus an async Path object, and code that runs unchanged on asyncio and trio |
| trio | PyPI | You are already on trio, where trio.open_file and trio.Path cover this without a third-party package |
| aiopath | PyPI | You want an async pathlib.Path replacement rather than an async open(), for code that mostly manipulates paths and metadata |