Real asynchronous file operations with asyncio support.

Features

Since version 2.0.0, uses caio, which provides multiple async I/O backends:
- Linux io_uring — the modern, high-performance Linux kernel AIO interface.
- Linux libaio — the classic kernel AIO mechanism via libaio.
- Thread-based (C) — a threadpool-backed implementation for POSIX systems (macOS, Linux).
- Pure Python — a thread-based fallback for any platform.
The best available backend is chosen automatically based on system compatibility.
AIOFile has no internal file pointer. Pass offset and chunk_size to each operation, or use the Reader/Writer helpers. For a file-like interface, use async_open.

Limitations

The Linux native AIO and io_uring backends cannot open special files. Asynchronous operations against special filesystems such as /proc/ or /sys/ are not supported by the kernel — this is neither an aiofile nor a caio issue. In such cases, switch to a thread-based implementation (see the Troubleshooting section).

Code examples

All code examples require Python 3.11+.

High-level API

`async_open` helper

This helper mimics Python's file-like objects, returning an object with equivalent asynchronous methods.

Supported methods:

async def read(length=-1) — reads a chunk from the file; -1 reads to the end.
async def write(data) — writes a chunk to the file.
def seek(offset) — sets the file pointer position.
def tell() — returns the current file pointer position.
async def readline(size=-1, newline="\n") — reads until a newline or EOF. Since version 3.7.0, __aiter__ returns a LineReader. This method is suboptimal for small lines because it does not reuse the read buffer — prefer LineReader when reading line by line.
def __aiter__() -> LineReader — iterator over lines.
def iter_chunked(chunk_size: int = 32768) -> Reader — iterator over chunks.
.file — the underlying AIOFile object.

Basic example:

import asyncio
from pathlib import Path
from tempfile import gettempdir
from aiofile import async_open
tmp_filename = Path(gettempdir()) / "hello.txt"
async def main():
    async with async_open(tmp_filename, 'w+') as afp:
        await afp.write("Hello ")
        await afp.write("world")
        afp.seek(0)
        print(await afp.read())
        await afp.write("Hello from\nasync world")
        print(await afp.readline())
        print(await afp.readline())
asyncio.run(main())

Example without context manager:

import asyncio
import atexit
import os
from tempfile import mktemp
from aiofile import async_open
TMP_NAME = mktemp()
atexit.register(os.unlink, TMP_NAME)
async def main():
    afp = await async_open(TMP_NAME, "w")
    await afp.write("Hello")
    await afp.close()
asyncio.run(main())
assert open(TMP_NAME, "r").read() == "Hello"

Concatenate example program (cat):

import asyncio
import sys
from argparse import ArgumentParser
from pathlib import Path
from aiofile import async_open
parser = ArgumentParser(
    description="Read files line by line using asynchronous io API"
)
parser.add_argument("file_name", nargs="+", type=Path)
async def main(arguments):
    for src in arguments.file_name:
        async with async_open(src, "r") as afp:
            async for line in afp:
                sys.stdout.write(line)
asyncio.run(main(parser.parse_args()))

Copy file example program (cp):

import asyncio
from argparse import ArgumentParser
from pathlib import Path
from aiofile import async_open
parser = ArgumentParser(
    description="Copying files using asynchronous io API"
)
parser.add_argument("source", type=Path)
parser.add_argument("dest", type=Path)
parser.add_argument("--chunk-size", type=int, default=65535)
async def main(arguments):
    async with async_open(arguments.source, "rb") as src, \
               async_open(arguments.dest, "wb") as dest:
        async for chunk in src.iter_chunked(arguments.chunk_size):
            await dest.write(chunk)
asyncio.run(main(parser.parse_args()))

Example with opening an already-open file pointer:

import asyncio
from typing import IO, Any
from aiofile import async_open
async def main(fp: IO[Any]):
    async with async_open(fp) as afp:
        await afp.write("Hello from\nasync world")
        print(await afp.readline())
with open("test.txt", "w+") as fp:
    asyncio.run(main(fp))

Linux native AIO and io_uring do not support reading or writing special files (procfs, sysfs, Unix pipes, etc.), so operations on these files require a compatible context object.

import asyncio
from aiofile import async_open
from caio import thread_aio_asyncio
from contextlib import AsyncExitStack
async def main():
    async with AsyncExitStack() as stack:
        # Custom context should be reused
        ctx = await stack.enter_async_context(
            thread_aio_asyncio.AsyncioContext()
        )
        # Open special file with custom context
        src = await stack.enter_async_context(
            async_open("/proc/cpuinfo", "r", context=ctx)
        )
        # Open regular file with default context
        dest = await stack.enter_async_context(
            async_open("/tmp/cpuinfo", "w")
        )
        # Copying file content line by line
        async for line in src:
            await dest.write(line)
asyncio.run(main())

`clone` helper

An asynchronous context supports a limited number of concurrent operations at the low level, regardless of how many file descriptors are open. clone lets you create a second file-like object with its own independent offset from a single descriptor, without opening the file multiple times.

"""
This example counts multiple hash functions from the file passed as the first
argument. The hash functions are counted competitively, and the results are
printed in the order of hashing completion.
"""
import asyncio
import hashlib
import sys
import aiofile
async def hasher(name, hash_func, afp):
    loop = asyncio.get_running_loop()
    async for chunk in afp.iter_chunked(2 ** 20):
        await loop.run_in_executor(None, hash_func.update, chunk)
    print(name, hash_func.hexdigest())
async def main():
    async with aiofile.async_open(sys.argv[1], "rb") as source:
        hashers = [
            ("MD5", hashlib.md5()),
            ("SHA1", hashlib.sha1()),
            ("SHA256", hashlib.sha256()),
            ("SHA512", hashlib.sha512()),
        ]
        await asyncio.gather(*[
            hasher(name, hash_func, await aiofile.clone(source))
            for name, hash_func in hashers
        ])
asyncio.run(main())

Note: This will likely perform poorly on Windows, so if that is your target platform, this optimization may not be worth it.

Low-level API

The AIOFile class is a low-level interface for asynchronous file operations. Its read and write methods accept an offset=0 (in bytes) at which the operation is performed.

This allows many independent I/O operations on a single open file without an internal pointer. For sequential reading and writing, use Writer, Reader, and LineReader. Note that async_open is not the same as AIOFile: it wraps it to provide a file-like interface similar to the built-in open.

import asyncio
from aiofile import AIOFile
async def main():
    async with AIOFile("hello.txt", 'w+') as afp:
        payload = "Hello world\n"
        await asyncio.gather(
            *[afp.write(payload, offset=i * len(payload)) for i in range(10)]
        )
        await afp.fsync()
        assert await afp.read(len(payload) * 10) == payload * 10
asyncio.run(main())

The low-level API is essentially a lightly sugared caio API.

import asyncio
from aiofile import AIOFile
async def main():
    async with AIOFile("/tmp/hello.txt", 'w+') as afp:
        await afp.write("Hello ")
        await afp.write("world", offset=7)
        await afp.fsync()
        print(await afp.read())
asyncio.run(main())

`Reader` and `Writer`

To read or write a file linearly, the following example may be helpful.

import asyncio
from aiofile import AIOFile, Reader, Writer
async def main():
    async with AIOFile("/tmp/hello.txt", 'w+') as afp:
        writer = Writer(afp)
        reader = Reader(afp, chunk_size=8)
        await writer("Hello")
        await writer(" ")
        await writer("World")
        await afp.fsync()
        async for chunk in reader:
            print(chunk)
asyncio.run(main())

`LineReader` - read file line by line

LineReader is a helper for reading a file linearly, line by line. It maintains a buffer and reads file fragments chunk by chunk, searching for line boundaries. The default chunk size is 4KB.

import asyncio
from aiofile import AIOFile, LineReader, Writer
async def main():
    async with AIOFile("/tmp/hello.txt", 'w+') as afp:
        writer = Writer(afp)
        await writer("Hello")
        await writer(" ")
        await writer("World")
        await writer("\n")
        await writer("\n")
        await writer("From async world")
        await afp.fsync()
        async for line in LineReader(afp):
            print(line)
asyncio.run(main())

To read a file line by line, prefer LineReader over async_open.

More examples

Async CSV Dict Reader

import asyncio
import io
from csv import DictReader
from aiofile import AIOFile, LineReader
class AsyncDictReader:
    def __init__(self, afp, **kwargs):
        self.buffer = io.BytesIO()
        self.file_reader = LineReader(
            afp, line_sep=kwargs.pop('line_sep', '\n'),
            chunk_size=kwargs.pop('chunk_size', 4096),
            offset=kwargs.pop('offset', 0),
        )
        self.reader = DictReader(
            io.TextIOWrapper(
                self.buffer,
                encoding=kwargs.pop('encoding', 'utf-8'),
                errors=kwargs.pop('errors', 'replace'),
            ), **kwargs,
        )
        self.line_num = 0
    def __aiter__(self):
        return self
    async def __anext__(self):
        if self.line_num == 0:
            header = await self.file_reader.readline()
            self.buffer.write(header)
        line = await self.file_reader.readline()
        if not line:
            raise StopAsyncIteration
        self.buffer.write(line)
        self.buffer.seek(0)
        try:
            result = next(self.reader)
        except StopIteration as e:
            raise StopAsyncIteration from e
        self.buffer.seek(0)
        self.buffer.truncate(0)
        self.line_num = self.reader.line_num
        return result
async def main():
    async with AIOFile('sample.csv', 'rb') as afp:
        async for item in AsyncDictReader(afp):
            print(item)
asyncio.run(main())

Troubleshooting

The caio Linux backends (io_uring and libaio) work well on modern kernels and filesystems. Problems are usually environment-specific and are not bugs. There are several ways to resolve them:

Upgrade the kernel.
Use a compatible filesystem.
Switch to a thread-based or pure-Python backend.

Since version 0.7.0, caio offers several ways to select the backend:

Set the CAIO_IMPL environment variable at runtime:
- uring — Linux io_uring (requires kernel ≥ 5.1)
- linux — Linux libaio
- thread — C-based thread pool implementation
- python — pure-Python thread-based fallback
The default_implementation file located next to __init__.py in the caio installation directory. Useful for distribution package maintainers. The file may contain comments (lines starting with #); the first non-comment line should be one of the values above.
Manage contexts manually:

import asyncio
from aiofile import async_open
from caio import linux_aio_asyncio, thread_aio_asyncio
async def main():
    linux_ctx = linux_aio_asyncio.AsyncioContext()
    threads_ctx = thread_aio_asyncio.AsyncioContext()
    async with async_open("/tmp/test.txt", "w", context=linux_ctx) as afp:
        await afp.write("Hello")
    async with async_open("/tmp/test.txt", "r", context=threads_ctx) as afp:
        print(await afp.read())
asyncio.run(main())

Benchmark

stdlib, aiofile (once per caio backend, each in its own subprocess), aiofiles, aiomisc.io — linear/random block order, read/write, buffered/O_DIRECT, sequential/concurrent, fixed op count per cell (never a partial/cancelled measurement). Tool, methodology and raw data: benchmark/; a third report at 4096B lives in benchmark/results/.

Linear write, one Linux/ext4 run, 65536B blocks, 2000 ops/cell, per-op latency in ms (p50 / p95, ±stdev):

participant	x1	x8
stdlib (buffered)	0.081 / 0.128 (±0.030)	0.167 / 0.298 (±0.157)
stdlib (O_DIRECT)	0.148 / 0.398 (±0.097)	0.266 / 0.650 (±0.192)
aiofile (linux_uring)	0.050 / 0.108 (±0.022)	0.040 / 0.248 (±0.293)
aiofile (linux_aio)	0.009 / 0.024 (±0.020)	0.040 / 0.089 (±0.020)
aiofile (thread_aio)	0.150 / 0.461 (±0.128)	0.664 / 2.930 (±0.943)
aiofile (python_aio)	0.109 / 0.173 (±0.042)	0.189 / 0.314 (±0.072)
aiofiles	0.155 / 0.257 (±0.175)	1.166 / 2.238 (±0.570)
aiomisc	0.156 / 0.256 (±0.051)	1.118 / 1.607 (±1.750)

linux_aio posts the lowest, tightest per-op latency at both concurrency levels here -- io_submit() is a lean, purpose-built syscall, while linux_uring's ring/SQPOLL machinery is built to amortize over batched submissions, not to minimize one op's latency, so it only pulls ahead once there's a batch to amortize over. aiofiles/aiomisc latency grows at x8 (one seek cursor, serialized by a lock) instead of shrinking like the positional-I/O libraries.

Linear write, 256B blocks, 2000 ops/cell, per-op latency in ms (p50 / p95, ±stdev):

participant	x1	x8
stdlib (buffered)	0.066 / 0.104 (±0.017)	0.152 / 0.299 (±0.125)
stdlib (O_DIRECT)	n/a (256B isn't 4096-aligned)	n/a
aiofile (linux_uring)	0.086 / 0.159 (±0.043)	0.043 / 0.070 (±0.021)
aiofile (linux_aio)	0.005 / 0.008 (±0.004)	0.017 / 0.027 (±0.008)
aiofile (thread_aio)	0.128 / 0.290 (±0.090)	0.464 / 1.503 (±0.470)
aiofile (python_aio)	0.113 / 0.201 (±0.037)	0.179 / 0.354 (±0.124)
aiofiles	0.136 / 0.169 (±0.030)	1.002 / 1.516 (±0.372)
aiomisc	0.144 / 0.223 (±0.040)	1.046 / 1.999 (±0.415)

Same shape at a tiny block size: linux_aio still has the lowest latency by a wide margin (sub-10μs median at x1), linux_uring still needs concurrency to close the gap (and still doesn't fully close it), and the lock-serialized libraries still get worse, not better, under concurrency.