Python aiofiles: How to Read & Write CSV Files Asynchronously

Introduction

This tutorial explores the use of aiofiles in Python for handling CSV files asynchronously, giving a boost to your I/O bound applications by not blocking the event loop during file operations.

Setting Up

First, ensure aiofiles is installed:

pip install aiofiles

For CSV operations, also ensure csv module is ready (it’s built into Python).

Basic Asynchronous File Reading

Begin by reading a CSV file asynchronously:

import aiofiles
import csv
import asyncio

async def read_csv(filepath):
    async with aiofiles.open(filepath, 'r') as file:
        reader = csv.reader(await file.read())
        for row in reader:
            print(row)

event_loop = asyncio.get_event_loop()
try:
    event_loop.run_until_complete(read_csv('yourfile.csv'))
finally:
    event_loop.close()

Asynchronous File Writing

To write to a CSV file asynchronously:

async def write_csv(filepath, data):
    async with aiofiles.open(filepath, 'w') as file:
        writer = csv.writer(file)
        await writer.writerow(['header1', 'header2'])
        for row in data:
            await writer.writerow(row)

Working with DictReader and DictWriter

Use DictReader and DictWriter for more control:

async def read_dict_csv(filepath):
    async with aiofiles.open(filepath, 'r') as file:
        content = await file.read()
        reader = csv.DictReader(content.splitlines())
        for row in reader:
            print(dict(row))

Advanced Usage

To handle large files, you can read/write in chunks:

async def read_large_csv(filepath):
    async with aiofiles.open(filepath, 'r') as file:
        while True:
            line = await file.readline()
            if not line:
                break
            print(csv.reader([line]).__next__())

Error Handling and Validation

Don’t forget to handle errors and validate data:

async def safe_read_csv(filepath):
    try:
        async with aiofiles.open(filepath, 'r') as file:
            content = await file.read()
            reader = csv.reader(content.splitlines())
            for row in reader:
                print(row)
    except FileNotFoundError:
        print(f"The file, {filepath}, does not exist.")
    except Exception as e:
        print(f"An error occurred: {str(e)}")

Integration with Async Web Frameworks

You might integrate with web frameworks like FastAPI to serve large CSV files without blocking.

from fastapi import FastAPI, Response
from fastapi.responses import StreamingResponse
import os

app = FastAPI()

@app.get("/")
async def stream_csv():
    path = 'huge_file.csv'
    file_size = os.path.getsize(path)

    def iterfile():
        with open(path, "rb") as file_like:
            yield from file_like

    response = StreamingResponse(iterfile(), media_type="text/csv")
    response.headers["Content-Length"] = str(file_size)
    return response

Conclusion

Achieving non-blocking I/O operations for CSV files in Python is straightforward with aiofiles. We covered reading and writing CSV files asynchronously and, for more advanced needs, how to handle large files without exhausting memory and integrating asynchronous file operations with web frameworks.