Python Trio循环中写入JSON的写法规范与优化咨询

你的代码功能上可以正常运行，但手动拼接JSON的分隔符与括号不够优雅，且同步文件IO会阻塞异步事件循环。以下是两种更符合Python风格的实现方案：

方案一：小数据量场景 - 先收集再写入 #

如果接收的数据量不大，可先将所有数据存入列表，再一次性写入JSON文件，代码更简洁易读：

import json
import aiofiles

async def rec(receiver):
    data = []
    async with receiver:
        async for val in receiver:
            data.append(val)
    
    print(f'[*] - Active items: {len(data)}')
    async with aiofiles.open('data.json', 'w', encoding='utf-8') as f:
        await f.write(json.dumps(data, indent=4, ensure_ascii=False))

这种方式避免了手动处理逗号和括号的麻烦，同时用aiofiles替代同步文件操作，不会阻塞异步事件循环。

方案二：大数据量场景 - 流式异步写入 #

若数据量过大无法一次性加载到内存，可优化流式写入的逻辑，同时使用异步文件IO：

import json
import aiofiles
from typing import AsyncIterable

async def rec(receiver: AsyncIterable[dict]):
    async with aiofiles.open('data.json', 'w', encoding='utf-8', buffering=1) as f:
        await f.write('[\n')
        count = 0
        async with receiver:
            # 先处理第一个元素，避免循环内重复判断
            iterator = receiver.__aiter__()
            try:
                first_val = await iterator.__anext__()
                count += 1
                await f.write(json.dumps(first_val, indent=4, ensure_ascii=False))
                # 处理剩余元素
                async for val in iterator:
                    count += 1
                    await f.write(',\n')
                    await f.write(json.dumps(val, indent=4, ensure_ascii=False))
            except StopAsyncIteration:
                pass
        print(f'[*] - Active items: {count}')
        await f.write('\n]')

关键优化点： #

• 使用aiofiles执行异步文件操作，这是异步函数中处理文件的标准做法，避免阻塞事件循环。
• 优化逗号处理逻辑，通过先取出第一个元素再遍历剩余元素的方式，减少循环内的条件判断次数，代码更高效清晰。
• 规范JSON格式的换行处理，避免手动拼接时出现格式错误。

内容的提问来源于stack exchange，提问作者αԋɱҽԃ αмєяιcαη