aiocsv

import asyncio import csv import aiofiles from aiocsv import AsyncReader, AsyncDictReader, AsyncWriter, AsyncDictWriter async def main(): # Create a dummy CSV file for reading async with aiofiles.open("example.csv", mode="w", encoding="utf-8", newline="") as afp: writer = AsyncWriter(afp) await writer.writerow(["name", "age"]) await writer.writerows([["John", 26], ["Sasha", 42]]) print("--- Reading simple CSV ---") async with aiofiles.open("example.csv", mode="r", encoding="utf-8", newline="") as afp: async for row in AsyncReader(afp): print(row) # Create a dummy TSV file for dict reading async with aiofiles.open("example.tsv", mode="w", encoding="utf-8", newline="") as afp: writer = AsyncDictWriter(afp, fieldnames=["name", "city"], delimiter="\t") await writer.writeheader() await writer.writerow({"name": "Alice", "city": "New York"}) await writer.writerow({"name": "Bob", "city": "London"}) print("\n--- Reading Dict CSV (TSV) ---") async with aiofiles.open("example.tsv", mode="r", encoding="utf-8", newline="") as afp: async for row in AsyncDictReader(afp, delimiter="\t"): print(row) # Writing new CSV file with AsyncWriter print("\n--- Writing simple CSV ---") async with aiofiles.open("output.csv", mode="w", encoding="utf-8", newline="") as afp: writer = AsyncWriter(afp, dialect="unix") await writer.writerow(["product", "price"]) await writer.writerows([["Laptop", 1200], ["Mouse", 25], ["Keyboard", 75]]) print("Written to output.csv") # Writing new CSV file with AsyncDictWriter print("\n--- Writing Dict CSV ---") async with aiofiles.open("output_dict.csv", mode="w", encoding="utf-8", newline="") as afp: writer = AsyncDictWriter(afp, ["item", "quantity", "status"], restval="N/A", quoting=csv.QUOTE_ALL) await writer.writeheader() await writer.writerow({"item": "Shirt", "quantity": 2, "status": "available"}) await writer.writerows([ {"item": "Pants", "quantity": 1}, {"item": "Socks", "quantity": 5, "status": "low stock"} ]) print("Written to output_dict.csv") asyncio.run(main())