Asynchronous File IO (caio)

Common errors

Warnings

Install compatibility verified last tested: 2026-05-12 v0.9.24 installed · v0.9.25 latest

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 last tested: 2026-04-24

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