如何在 FastAPI 中读取 UploadFile？

Question

我有一个 FastAPI 端点，它接收一个文件，将它上传到 s3，然后处理它。一切正常，除了处理失败并显示此消息：

  File "/usr/local/lib/python3.9/site-packages/starlette/datastructures.py", line 441, in read
    return self.file.read(size)
  File "/usr/local/lib/python3.9/tempfile.py", line 735, in read
    return self._file.read(*args)
ValueError: I/O operation on closed file.

我的简化代码如下所示：

async def process(file: UploadFile):
    reader = csv.reader(iterdecode(file.file.read(), "utf-8"), dialect="excel")  # This fails!
    datarows = []
    for row in reader:
        datarows.append(row)
    return datarows

如何读取上传文件的内容？

更新

我设法进一步隔离了问题。这是我的简化端点：

import boto3
from loguru import logger
from botocore.exceptions import ClientError


UPLOAD = True

@router.post("/")
async def upload(file: UploadFile = File(...)):
    if UPLOAD:
        # Upload the file
        s3_client = boto3.client("s3", endpoint_url="http://localstack:4566")
        try:
            s3_client.upload_fileobj(file.file, "local", "myfile.txt")
        except ClientError as e:
            logger.error(e)
    contents = await file.read()
    return JSONResponse({"message": "Success!"})

如果 UPLOAD 为 True，我会收到错误消息。如果不是，一切正常。上传后似乎 boto3 正在关闭文件。有什么办法可以重新打开文件吗？或者发一份给upload_fileobj？

Answer 1

根据FastAPI's documentation，UploadFile 使用Python 的SpooledTemporaryFile，“存储在内存中的文件达到最大大小限制，超过此限制后它将存储在磁盘中。”。它“与TemporaryFile 完全一样运行”，它“在关闭后立即被销毁（包括对象被垃圾回收时的隐式关闭）”。似乎，一旦 boto3 读取了文件的内容，文件就会关闭，进而导致文件被删除。

选项 1

像你已经做的那样读取文件内容（即contents = await file.read()），然后将这些字节上传到你的服务器，而不是文件对象（如果可能的话）。

更新

我还应该提到，有一个 表示位置的内部“光标”（或“文件指针”） 文件内容将被读取（或写入）。打电话时 read() 一直读取到缓冲区的末尾，留下零 超出光标的字节。因此，也可以使用seek() 方法 将光标的当前位置设置为 0（即，倒回 光标到文件的开头）；因此，允许您在读取内容后传递文件对象。根据FastAPI's documentation：

seek(offset)：转到文件中的字节位置offset (int)。

• 例如，await myfile.seek(0) 将转到文件的开头。
• 如果您运行await myfile.read() 一次然后需要再次读取内容，这将特别有用。

选项 2

将文件内容复制到NamedTemporaryFile，与TemporaryFile不同，它“在文件系统中有一个可见的名称”，“可以用来打开文件”。此外，通过将 delete 参数设置为 False，它在关闭后仍可访问；因此，允许文件在需要时重新打开。完成后，您可以使用 remove() 或 unlink() 方法手动删除它。下面是一个工作示例（灵感来自this answer）：

import uvicorn
from fastapi import FastAPI, File, UploadFile
from tempfile import NamedTemporaryFile
import os


app = FastAPI()

@app.post("/uploadfile/")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()

    file_copy = NamedTemporaryFile(delete=False)
    try:
        file_copy.write(contents);  # copy the received file data into a new temp file. 
        file_copy.seek(0)  # move to the beginning of the file
        print(file_copy.read(10))
        
        # Here, upload the file to your S3 service

    finally:
        file_copy.close()  # Remember to close any file instances before removing the temp file
        os.unlink(file_copy.name)  # unlink (remove) the file
    
    # print(contents)  # Handle file contents as desired
    return {"filename": file.filename}
    

if __name__ == '__main__':
    uvicorn.run(app, host='127.0.0.1', port=8000, debug=True)

更新

如果文件需要在关闭之前重新打开（当它被上传到您的服务器时），并且您的平台不允许这样做（如here 所述），请使用以下代码代替（临时文件的使用file_copy.name访问路径）：

@app.post("/uploadfile/")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()

    file_copy = NamedTemporaryFile('wb', delete=False)
    f = None
    try:
        # The 'with' block ensures that the file closes and data are stored
        with file_copy as f:
            f.write(contents);
        
        # Here, upload the file to your S3 service
        # You can reopen the file as many times as desired. 
        f = open(file_copy.name, 'rb')
        print(f.read(10))

    finally:
        if f is not None:
            f.close() # Remember to close any file instances before removing the temp file
        os.unlink(file_copy.name)  # unlink (remove) the file from the system's Temp folder
    
    # Handle file contents as desired
    # print(contents)
    return {"filename": file.filename}

选项 3

您甚至可以将字节保存在内存缓冲区BytesIO 中，使用它将内容上传到 S3 存储桶，最后关闭它（“当close() 方法被丢弃时，缓冲区将被丢弃称为。”）。写入 BytesIO 流后记得调用seek(0) 方法将光标重置回文件的开头。

contents = await file.read()
temp_file = io.BytesIO()
temp_file.write(contents)
temp_file.seek(0)
s3_client.upload_fileobj(temp_file, "local", "myfile.txt")
temp_file.close()

Answer 2

来自 FastAPI ImportFile：

从 fastapi 导入文件和 UploadFile：

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/files/")
async def create_file(file: bytes = File(...)):
    return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

来自 FastAPI UploadFile：

例如，在异步路径操作函数中，您可以获得 内容：

contents = await myfile.read()

你的代码应该是这样的：

async def process(file: UploadFile = File(...)):
    content = await file.read()
    reader = csv.reader(iterdecode(content, "utf-8"), dialect="excel")
    datarows = []
    for row in reader:
        datarows.append(row)
    return datarows