如何从磁盘读取JPEG并实例化可正常保存的werkzeug FileStorage？

环境 #

python  3.10.11
werkzeug  2.3.7

我有一张JPEG图片位于/mnt/dragon.jpeg，需要为一个接收werkzeug.datastructures.FileStorage作为输入、并将其保存到指定路径的方法编写单元测试。

我尝试用以下代码实例化FileStorage：

import os
from werkzeug.datastructures import FileStorage

PATH = "/mnt/dragon.jpeg"

def get_file(filepath: str) -> FileStorage:
    filename = os.path.basename(filepath)
    with open(filepath, 'rb') as file:
        return FileStorage(file, name=filename)

file = get_file(PATH)

在IPython中执行时出现报错：

In [8]: file
Out[8]: <FileStorage: '/mnt/dragon.jpeg' (None)>

In [9]: file.save("/mnt/saved.jpeg")
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
Cell In[9], line 1
----> 1 file.save("/mnt/saved.jpeg")

File ~/.local/lib/python3.10/site-packages/werkzeug/datastructures/file_storage.py:129, in FileStorage.save(self, dst, buffer_size)
    126     close_dst = True
    128 try:
--> 129     copyfileobj(self.stream, dst, buffer_size)
    130 finally:
    131     if close_dst:

File ~/miniconda3/lib/python3.10/shutil.py:195, in copyfileobj(fsrc, fdst, length)
    193 fdst_write = fdst.write
    194 while True:
--> 195     buf = fsrc_read(length)
    196     if not buf:
    197         break

ValueError: read of closed file

请问如何让实例化后的werkzeug.datastructures.FileStorage能在open语句的作用域外正常保存？

解决方案 #

问题根源是with open的特性：代码块执行完毕后，文件流会自动关闭，而返回的FileStorage对象持有的是已关闭的流，后续调用save时自然无法读取。

以下是两种可行的修复方式：

方式1：手动管理文件流（需注意后续关闭） #

放弃with open，直接打开文件并将流交给FileStorage，保存完成后手动关闭流：

import os
from werkzeug.datastructures import FileStorage

PATH = "/mnt/dragon.jpeg"

def get_file(filepath: str) -> FileStorage:
    filename = os.path.basename(filepath)
    file = open(filepath, 'rb')
    return FileStorage(file, name=filename)

file = get_file(PATH)
try:
    file.save("/mnt/saved.jpeg")
finally:
    file.stream.close()

方式2：将文件内容加载到内存流（推荐，小文件适用） #

先读取文件全部内容到内存，用BytesIO包装成可读取的流，再传给FileStorage，这样就不受原文件流关闭的影响：

import os
from io import BytesIO
from werkzeug.datastructures import FileStorage

PATH = "/mnt/dragon.jpeg"

def get_file(filepath: str) -> FileStorage:
    filename = os.path.basename(filepath)
    with open(filepath, 'rb') as f:
        content = f.read()
    return FileStorage(BytesIO(content), name=filename)

file = get_file(PATH)
file.save("/mnt/saved.jpeg")

补充：让FileStorage自动接管流关闭 #

如果你不想手动关闭流，可以直接打开文件并传给FileStorage，它默认会在save方法执行完成后关闭流：

def get_file(filepath: str) -> FileStorage:
    filename = os.path.basename(filepath)
    return FileStorage(open(filepath, 'rb'), name=filename)

file = get_file(PATH)
file.save("/mnt/saved.jpeg")  # 执行后流会自动关闭

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