aiofile Documentation

If you encounter issues with caio Linux backends (io_uring or libaio), they are typically environment-specific rather than bugs. You can resolve these by upgrading your kernel, using a compatible filesystem, or switching to a different backend.

To switch backends, you can use one of the following methods:

import asyncio
from aiofile import async_open
from caio import linux_aio_asyncio, thread_aio_asyncio

async def main():
    # Create specific contexts
    linux_ctx = linux_aio_asyncio.AsyncioContext()
    threads_ctx = thread_aio_asyncio.AsyncioContext()

    # Use the linux context for writing
    async with async_open("/tmp/test.txt", "w", context=linux_ctx) as afp:
        await afp.write("Hello")

    # Use the thread context for reading
    async with async_open("/tmp/test.txt", "r", context=threads_ctx) as afp:
        print(await afp.read())

asyncio.run(main())

Since AIOFile has no internal pointer, use the Reader and Writer helpers to perform sequential (linear) I/O operations.

• Writer(afp): Wraps an AIOFile to provide sequential writing.
• Reader(afp, chunk_size=...): Wraps an AIOFile to provide sequential reading in chunks.

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())

To ensure a specific backend is used for a file operation, you can manually create a context from the caio package and pass it to aiofile.async_open using the context parameter. This bypasss the default implementation selection.

from aiofile import async_open
from caio import linux_aio_asyncio, thread_aio_asyncio

# Example: Using a specific Linux AIO context
linux_ctx = linux_aio_asyncio.AsyncioContext()
async with async_open("filename.txt", "r", context=linux_ctx) as afp:
    content = await afp.read()

# Example: Using a thread-based context
threads_ctx = thread_aio_asyncio.AsyncioContext()
async with async_open("filename.txt", "r", context=threads_ctx) as afp:
    content = await afp.read()

The AIOFile class is a low-level interface where operations do not rely on an internal file pointer. Instead, you must pass an offset (in bytes) to read and write methods. This allows for highly efficient, independent I/O operations at specific locations in the file.

Key Methods:

• await afp.read(length, offset=0)
• await afp.write(data, offset=0)
• await afp.fsync()

import asyncio
from aiofile import AIOFile

async def main():
    async with AIOFile("hello.txt", 'w+') as afp:
        payload = "Hello world\n"

        # Perform multiple writes at different offsets concurrently
        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())

For efficient line-by-line reading, use the LineReader helper. It maintains an internal buffer and reads file fragments in chunks (default 4KB) to search for line boundaries. This is more performant than using async_open().readline() for small lines because it reuses the read buffer.

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\n")
        await writer("World\n")
        await writer("From async world")
        await afp.fsync()

        async for line in LineReader(afp):
            print(line)

asyncio.run(main())

You can control which backend caio uses by setting the CAIO_IMPL environment variable at runtime. This is useful for switching between high-performance Linux backends and more compatible thread-based fallbacks without changing code.

# Available CAIO_IMPL values:
# uring  -> Linux io_uring (requires kernel ≥ 5.1)
# linux  -> Linux libaio
# thread -> C-based thread pool implementation
# python -> pure-Python thread-based fallback

The clone helper function allows you to create a clone of a file object, which can be useful for duplicating file handles or state.

from aiofile import clone

# Assuming 'f' is an existing file object
new_f = clone(f)

Use async_open() to obtain a high-level file wrapper that supports both binary and text modes. If the mode is not binary, it returns a TextFileWrapper; otherwise, it returns a BinaryFileWrapper. The returned wrapper can be used as an asynchronous context manager.

Arguments:

• file_specifier: A str, Path, or FileIOType object.
• mode: The file mode (e.g., 'r', 'w', 'rb', 'wb'). Defaults to 'r'.
• *args, **kwargs: Passed to the underlying AIOFile constructor.

from aiofile.utils import async_open

# For text mode
async with async_open("example.txt", mode="r") as f:
    content = await f.read()
    print(content)

# For binary mode
async with async_open("example.bin", mode="rb") as f:
    data = await f.read(1024)