aiocsv_async_csv_read_write_with_dictwriter.py

1import asyncio
2import csv
3import aiofiles
4from aiocsv import AsyncReader, AsyncDictReader, AsyncWriter, AsyncDictWriter
5
6async def main():
7    # Reading
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    # DictReading
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
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", 18], ["bob", 21]])
22
23    # DictWriting
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": 18})
28        await writer.writerows([{"name": "bob", "age": 21}, {"name": "charlie", "age": 45}])
29
30if __name__ == "__main__":
31    asyncio.run(main())