aiocsv_async_csv_read_write_with_aiofiles.py

1import asyncio
2import csv
3import aiofiles
4from aiocsv import AsyncReader, AsyncDictReader, AsyncWriter, AsyncDictWriter
5
6async def main():
7    # Reading from a csv file
8    async with aiofiles.open("test.csv", mode="r", encoding="utf-8", newline="") as afp:
9        async for row in AsyncReader(afp):
10            print(row)  # row is a list
11
12    # Reading from a csv file using DictReader
13    async with aiofiles.open("test.csv", mode="r", encoding="utf-8", newline="") as afp:
14        async for row in AsyncDictReader(afp):
15            print(row)  # row is a dict
16
17    # Writing to a csv file
18    async with aiofiles.open("test.csv", mode="w", encoding="utf-8", newline="") as afp:
19        writer = AsyncWriter(afp)
20        await writer.writerow(["name", "age"])
21        await writer.writerows([["alice", 25], ["bob", 30]])
22
23    # Writing to a csv file using DictWriter
24    async with aiofiles.open("test.csv", mode="w", encoding="utf-8", newline="") as afp:
25        writer = AsyncDictWriter(afp, ["name", "age"])
26        await writer.writeheader()
27        await writer.writerow({"name": "alice", "age": 25})
28        await writer.writerows([{"name": "bob", "age": 30}, {"name": "charlie", "age": 35}])
29
30if __name__ == "__main__":
31    asyncio.run(main())