Python async-await函数内使用JSON/pickle读写文件问题咨询

在采用async-await异步模式的discord.py框架下开发时，需要通过pickle、JSON模块完成数据序列化写入（dump）与反序列化加载（load），执行相关操作时最初抛出如下报错：

AttributeError: __enter__
d:\Users\-------\visual studio projects\------------\main.py:65: RuntimeWarning: 
coroutine 'Command.__call__' was never awaited

最初判断问题来自异步函数内使用同步方式打开文件，因此尝试改用aiofiles库异步打开文件，测试代码如下：

async with aiofiles.open("owners.pkl", mode="rb") as file:
    owner_dict = pickle.load(file)

测试发现pickle、json模块无法直接配合异步文件对象在async函数内正常工作，需要可落地的实现方案。

报错原因说明：

• AttributeError: __enter__ 是因为异步上下文管理器（aiofiles.open返回的对象）只能搭配async with使用，如果误写为同步with就会找不到同步上下文管理器要求的__enter__方法；异常抛出后命令协程未被正常await，就会连带出现coroutine 'Command.__call__' was never awaited警告。
• 直接在异步文件对象上调用pickle/json的load方法失效，是因为这两个模块的load方法会同步调用文件对象的read接口，而异步文件对象的read是需要await的协程，无法被同步序列化逻辑直接调用。

方案1：通过asyncio线程池执行阻塞IO（优先推荐） #

不需要强行适配异步文件对象，直接把阻塞的文件读写、序列化/反序列化逻辑放到独立线程执行，既不会阻塞asyncio事件循环，也完全兼容原生pickle、json的所有语法，无额外适配成本，也不受文件体积限制。

工具函数实现 #

import asyncio
import pickle
import json
from typing import Any

# 反序列化加载
async def load_pickle(file_path: str) -> Any:
    def _sync_load():
        with open(file_path, "rb") as f:
            return pickle.load(f)
    return await asyncio.to_thread(_sync_load)

async def load_json(file_path: str, encoding: str = "utf-8") -> Any:
    def _sync_load():
        with open(file_path, "r", encoding=encoding) as f:
            return json.load(f)
    return await asyncio.to_thread(_sync_load)

# 序列化写入
async def dump_pickle(data: Any, file_path: str) -> None:
    def _sync_dump():
        with open(file_path, "wb") as f:
            pickle.dump(data, f)
    await asyncio.to_thread(_sync_dump)

async def dump_json(data: Any, file_path: str, encoding: str = "utf-8", indent: int = 2, ensure_ascii: bool = False) -> None:
    def _sync_dump():
        with open(file_path, "w", encoding=encoding) as f:
            json.dump(data, f, indent=indent, ensure_ascii=ensure_ascii)
    await asyncio.to_thread(_sync_dump)

调用示例 #

# 读取
owner_dict = await load_pickle("owners.pkl")
config = await load_json("config.json")

# 写入
await dump_pickle(owner_dict, "owners.pkl")
await dump_json(config, "config.json")

注意：Python 3.9及以上版本原生支持asyncio.to_thread，如果使用更低版本Python，将await asyncio.to_thread(xxx)替换为await asyncio.get_event_loop().run_in_executor(None, xxx)即可实现相同效果。

方案2：aiofiles + 内存序列化/反序列化 #

如果一定要使用aiofiles，可以先通过aiofiles异步把整个文件内容读取到内存，再用pickle.loads/json.loads处理内存中的数据；写入时先在内存中完成序列化，再通过aiofiles异步写入文件即可。

代码示例 #

import aiofiles
import pickle
import json

# 读取pickle
async with aiofiles.open("owners.pkl", "rb") as f:
    file_content = await f.read()
owner_dict = pickle.loads(file_content)

# 写入pickle
async with aiofiles.open("owners.pkl", "wb") as f:
    file_content = pickle.dumps(owner_dict)
    await f.write(file_content)

# 读取json
async with aiofiles.open("config.json", "r", encoding="utf-8") as f:
    file_content = await f.read()
config = json.loads(file_content)

# 写入json
async with aiofiles.open("config.json", "w", encoding="utf-8") as f:
    file_content = json.dumps(config, indent=2, ensure_ascii=False)
    await f.write(file_content)

注意：该方案会一次性把整个文件内容加载到内存，仅适合文件体积较小的场景，大文件操作优先选择方案1。

内容的提问来源于stack exchange，提问作者Zaid abuelbeh