Asynchronous File I/O with Aiofiles in Asyncio

Python's built-in open() function is blocking — it pauses the entire event loop during file operations, freezing concurrent tasks. The aiofiles library solves this by providing an async-compatible file API that integrates seamlessly with asyncio, allowing file reads/writes to happen without blocking other coroutines.

Key Takeaways

• Standard open() blocks the entire event loop — file operations are slow relative to CPU execution, so blocking during I/O defeats the purpose of async
• aiofiles.open() is an async context manager — use async with aiofiles.open(...) to safely open and close files without blocking
• All file methods are coroutines — .read(), .write(), .readlines() must be awaited; awaiting them yields control back to the event loop
• Install with pip install aiofiles — it's the standard third-party solution for async file operations in the Python community

---

Why Is Standard Python File I/O Blocking?

Python's built-in open() and file operations like .read() and .write() are synchronous and blocking. They pause execution in the current thread until the entire operation completes. In an asyncio application, a thread is shared by all running coroutines. When one coroutine blocks on file I/O, it stalls the entire event loop, preventing other coroutines from running.

# BLOCKING example — this is bad in asyncio
import asyncio

async def read_file_blocking():
    """This blocks the event loop."""
    # open() is synchronous and blocking
    with open('large_file.txt', 'r') as f:
        data = f.read()  # Blocks until entire file is read
    return data

async def other_task():
    """This cannot run until read_file_blocking completes."""
    for i in range(5):
        print(f"Task {i}")
        await asyncio.sleep(0.1)

async def main():
    # Both tasks run in order, NOT concurrently
    await read_file_blocking()  # Entire event loop waits here
    await other_task()  # Runs only after file operation finishes

# This is inefficient — file I/O and task execution are serialized
asyncio.run(main())

The aiofiles library solves this by delegating file operations to a thread pool, freeing the event loop to run other coroutines while waiting.

---

How Do You Use Aiofiles to Read and Write Files Asynchronously?

aiofiles provides an async-compatible file API that mirrors the built-in open(). The key difference is that all operations are coroutines and must be used with async/await.

Installing Aiofiles

pip install aiofiles

Reading Files Asynchronously

import asyncio
import aiofiles

async def read_file_async():
    """Read a file without blocking the event loop."""
    # Use async with to open the file
    async with aiofiles.open('data.txt', mode='r') as f:
        # .read() is a coroutine — must be awaited
        contents = await f.read()
    
    print(contents)
    return contents

asyncio.run(read_file_async())

Writing Files Asynchronously

import asyncio
import aiofiles

async def write_file_async():
    """Write to a file without blocking the event loop."""
    async with aiofiles.open('output.txt', mode='w') as f:
        # .write() is a coroutine — must be awaited
        await f.write("Line 1: Hello from aiofiles!\n")
        await f.write("Line 2: This is non-blocking I/O.\n")
        await f.write("Line 3: Multiple tasks can run concurrently.\n")

asyncio.run(write_file_async())

Reading Line-by-Line Asynchronously

For large files, read line-by-line to avoid loading everything into memory:

import asyncio
import aiofiles

async def read_lines_async():
    """Read a file line-by-line without blocking."""
    async with aiofiles.open('large_file.txt', mode='r') as f:
        # Async iteration over lines
        async for line in f:
            print(f"Processing: {line.strip()}")

asyncio.run(read_lines_async())

Appending to a File

import asyncio
import aiofiles

async def append_to_file():
    """Append data to a file asynchronously."""
    async with aiofiles.open('log.txt', mode='a') as f:
        await f.write("New log entry\n")

asyncio.run(append_to_file())

---

How Can You Run File Operations Concurrently With Other Tasks?

The real power of aiofiles is running multiple I/O-bound tasks concurrently. While one coroutine waits on file I/O, the event loop executes other coroutines.

Concurrent File Operations and Background Tasks

import asyncio
import aiofiles
import time

async def write_lines_async():
    """Simulate a slow file write operation."""
    print("[Write] Starting to write file...")
    async with aiofiles.open('concurrent_demo.txt', mode='w') as f:
        for i in range(5):
            await f.write(f"Line {i+1}\n")
            # Simulate I/O delay
            await asyncio.sleep(0.2)
    print("[Write] File write complete.")

async def background_counter():
    """A background task that runs while file I/O is happening."""
    print("[Counter] Starting background counter...")
    for i in range(1, 6):
        await asyncio.sleep(0.2)
        print(f"[Counter] Count: {i}")
    print("[Counter] Background counter done.")

async def main():
    """Run both tasks concurrently."""
    start = time.time()
    
    # Both tasks run at the same time
    await asyncio.gather(
        write_lines_async(),
        background_counter()
    )
    
    elapsed = time.time() - start
    print(f"\nBoth tasks completed in {elapsed:.2f} seconds (concurrent)")
    # With aiofiles: ~1.0 seconds
    # With standard file I/O: ~2.0 seconds (sequential)

asyncio.run(main())

Output:

[Write] Starting to write file...
[Counter] Starting background counter...
[Counter] Count: 1
[Write] File write complete.
[Counter] Count: 2
[Counter] Count: 3
[Counter] Count: 4
[Counter] Count: 5
[Counter] Background counter done.

Both tasks completed in 1.01 seconds (concurrent)

Without aiofiles, the counter would be frozen during file writes, and total time would be ~2 seconds instead of 1.

Reading Multiple Files Concurrently

import asyncio
import aiofiles

async def read_file(filename):
    """Read a single file and return its contents."""
    async with aiofiles.open(filename, mode='r') as f:
        contents = await f.read()
    return (filename, contents)

async def read_multiple_files(filenames):
    """Read multiple files concurrently."""
    # Create a task for each file
    tasks = [read_file(filename) for filename in filenames]
    # Run all tasks concurrently
    results = await asyncio.gather(*tasks)
    return results

async def main():
    files = ['file1.txt', 'file2.txt', 'file3.txt']
    results = await read_multiple_files(files)
    
    for filename, contents in results:
        print(f"--- {filename} ---")
        print(contents)
        print()

asyncio.run(main())

Reading three files concurrently is much faster than reading them sequentially.

---

What Are the Common File Modes and Methods in Aiofiles?

aiofiles.open() supports the same modes and methods as standard Python open():

Common Modes

Mode	Purpose
'r'	Read (default)
'w'	Write (truncates if file exists)
'a'	Append to end of file
'rb'	Read binary
'wb'	Write binary

Common Methods (All Async Coroutines)

async with aiofiles.open('file.txt', 'r') as f:
    # All of these must be awaited
    
    content = await f.read()  # Read entire file into memory
    line = await f.readline()  # Read one line
    lines = await f.readlines()  # Read all lines into a list
    await f.write("text")  # Write text
    await f.seek(0)  # Move file position
    position = await f.tell()  # Get current file position
    await f.flush()  # Flush buffer to disk

---

Frequently Asked Questions

Can you use standard open() with asyncio?

Technically yes, but it's bad practice. Standard open() blocks the event loop, defeating the purpose of async/await. If you must use blocking I/O, run it in a thread pool using loop.run_in_executor(), but aiofiles is cleaner and more idiomatic.

Does aiofiles work with binary files?

Yes, use 'rb' or 'wb' modes:

async with aiofiles.open('image.png', mode='rb') as f:
    binary_data = await f.read()

Is aiofiles thread-safe?

aiofiles itself is designed for use within a single event loop (not thread-safe across threads). However, you can safely use it in multi-threaded applications by ensuring each thread has its own event loop.

What if the file doesn't exist when reading?

aiofiles raises the same exceptions as standard Python — FileNotFoundError when reading a non-existent file. Wrap operations in try/except as needed:

import asyncio
import aiofiles

async def safe_read(filename):
    try:
        async with aiofiles.open(filename, 'r') as f:
            return await f.read()
    except FileNotFoundError:
        print(f"{filename} not found")
        return None

asyncio.run(safe_read('missing.txt'))

How does aiofiles handle large files?

For large files, iterate line-by-line with async for instead of reading the entire file into memory:

async with aiofiles.open('huge_file.txt', 'r') as f:
    async for line in f:
        # Process each line without loading the entire file
        print(line.strip())

---

Further Reading

• Aiofiles Official Documentation
• Python Official Docs: asyncio
• Real Python: Async I/O in Python