如何将Strawberry API中的SpooledTemporaryFile写入磁盘?
如何在Strawberry API中将上传的文件写入磁盘?
你当前的Strawberry API代码如下:
import typing import strawberry from strawberry.file_uploads import Upload @strawberry.input class FolderInput: files: typing.List[Upload] @strawberry.type class Mutation: @strawberry.mutation async def read_file(self, file: Upload) -> str: return (await file.read()).decode(...)
Strawberry的Upload对象底层对应类似SpooledTemporaryFile的文件对象,要将其写入磁盘,有两种高效实现方式:
方式一:读取文件内容后写入磁盘
这种方式适合小文件,直接读取字节内容后写入本地文件:
import typing import strawberry import os from strawberry.file_uploads import Upload @strawberry.input class FolderInput: files: typing.List[Upload] @strawberry.type class Mutation: @strawberry.mutation async def save_file(self, file: Upload) -> str: # 读取上传文件的字节内容 file_content = await file.read() # 定义保存路径,可根据业务调整(这里用原始文件名) save_path = f"./uploaded_files/{file.filename}" # 自动创建不存在的目录 os.makedirs(os.path.dirname(save_path), exist_ok=True) # 写入磁盘 with open(save_path, "wb") as local_file: local_file.write(file_content) return f"文件 {file.filename} 已保存至 {save_path}"
方式二:用shutil复制文件对象(大文件更高效)
对于大文件,直接复制文件对象能避免一次性加载全部内容到内存,更节省资源:
import typing import strawberry import shutil import os from strawberry.file_uploads import Upload @strawberry.input class FolderInput: files: typing.List[Upload] @strawberry.type class Mutation: @strawberry.mutation async def save_file(self, file: Upload) -> str: save_path = f"./uploaded_files/{file.filename}" os.makedirs(os.path.dirname(save_path), exist_ok=True) # 获取底层文件对象 temp_file = await file.open() try: # 将文件指针移至开头(防止之前读取过导致内容为空) temp_file.seek(0) # 复制文件对象到本地文件 with open(save_path, "wb") as local_file: shutil.copyfileobj(temp_file, local_file) finally: # 确保临时文件关闭 await file.close() return f"文件 {file.filename} 已保存至 {save_path}"
注意事项
- 保存路径中的
file.filename是上传时的原始文件名,可能存在重名,建议添加唯一标识(如时间戳、UUID)避免文件覆盖 - 处理多文件时,可遍历
FolderInput中的files列表,对每个Upload对象执行上述保存逻辑 - 必须确保保存目录存在,
os.makedirs(..., exist_ok=True)会自动创建缺失的目录
内容的提问来源于stack exchange,提问作者easythrees
相关产品推荐
相关产品推荐

