Asynchronous File IO (caio)

0.9.25 · active · verified Wed Apr 01

caio is a Python library providing asynchronous file I/O for Linux, macOS, and Windows. It offers Python bindings for Linux AIO API, including `io_uring`, and provides fallback mechanisms for other platforms using threads or pure Python. Currently at version 0.9.25, the library is actively maintained with regular updates.

Warnings

gotcha The `io_uring` backend might be blocked by `seccomp` filters in container environments like Docker, Podman, or Kubernetes. This can lead to `ImportError` when `io_uring_setup(2)` returns `ENOSYS`.
Fix: To fix this, run containers with `--security-opt seccomp=unconfined` (Docker/Podman) or set `securityContext.seccompProfile.type: Unconfined` (Kubernetes).
gotcha Native Linux AIO implementation requires a kernel version of 4.18 or newer. If an older kernel is detected, `caio` will fall back to a thread-based or pure Python implementation, which might have different performance characteristics.
Fix: Ensure your Linux kernel is 4.18 or newer for optimal performance with native AIO. Alternatively, explicitly select a backend using the `CAIO_IMPL` environment variable (`CAIO_IMPL=thread` or `CAIO_IMPL=python`) or by importing a specific backend directly (e.g., `from caio.thread_aio_asyncio import AsyncioContext`).
deprecated Direct imports of specific backend implementations (e.g., `from caio.linux_aio_asyncio import AsyncioContext`) were previously the primary way to force a backend. While still possible, it is now recommended to let `caio` pick the best available backend automatically via `from caio import AsyncioContext` or to use the `CAIO_IMPL` environment variable or a `default_implementation` file for global control.
Fix: Use `from caio import AsyncioContext` for automatic backend selection. For explicit control, set the `CAIO_IMPL` environment variable (e.g., `CAIO_IMPL=uring`) or create a `default_implementation` file with the desired backend name (e.g., `uring`).

Install

pip install caio Install stable version

Imports

AsyncioContext
```
from caio import AsyncioContext
```
This import automatically selects the best available backend (linux_uring → linux_aio → thread_aio → python_aio).
AsyncioContext (linux_uring)
```
from caio.linux_uring_asyncio import AsyncioContext
```
Use this to explicitly force the `io_uring` backend on Linux kernels >= 5.6.
AsyncioContext (linux_aio)
```
from caio.linux_aio_asyncio import AsyncioContext
```
Use this to explicitly force the `linux_aio` backend on Linux kernels >= 4.18.
AsyncioContext (thread)
```
from caio.thread_aio_asyncio import AsyncioContext
```
Use this to explicitly force the thread-based backend, which is portable.
AsyncioContext (python)
```
from caio.python_aio_asyncio import AsyncioContext
```
Use this to explicitly force the pure Python backend, which requires no C extension.

Quickstart

This quickstart demonstrates how to use `caio` with `asyncio` to perform basic asynchronous file operations like writing, reading, and synchronizing. It also shows how to execute multiple write operations concurrently. A temporary file is created and cleaned up for the demonstration.

import asyncio
import os
from caio import AsyncioContext

async def main():
    # Ensure a dummy file exists for the example
    file_path = "test.file"
    with open(file_path, "wb+") as f: # Create or truncate the file
        f.write(b"")

    ctx = AsyncioContext(max_requests=128)
    fd = os.open(file_path, os.O_RDWR | os.O_CREAT)

    try:
        # Execute one write operation
        await ctx.write(b"Hello world", fd, offset=0)
        print(f"Wrote: Hello world")

        # Execute one read operation
        read_data = await ctx.read(32, fd, offset=0)
        print(f"Read: {read_data.decode()}")

        # Execute one fdsync operation
        await ctx.fdsync(fd)
        print("File synchronized.")

        # Execute multiple writes concurrently
        op1 = ctx.write(b"Hello from ", fd, offset=0)
        op2 = ctx.write(b"async world", fd, offset=11)
        await asyncio.gather(op1, op2)
        print("Concurrent writes completed.")

        read_data_concurrent = await ctx.read(32, fd, offset=0)
        print(f"Read after concurrent writes: {read_data_concurrent.decode()}")

    finally:
        os.close(fd)
        os.remove(file_path)

if __name__ == '__main__':
    asyncio.run(main())

view raw JSON →