python - 如何在 FastAPI 中上传 CSV 文件并将其转换为 JSON？

Question

我正在尝试将我的 .csv 文件上传到我的 FastAPI 服务器，然后将其转换为 JSON 并将其返回给客户端。但是，当我尝试直接处理它(而不将其存储在某处)时，出现此错误:

Error : FileNotFoundError: [Error 2] No such file or directory : "testdata.csv"

这是我的 FastAPI 代码:

async def upload(file: UploadFile = File(...)):
    data = {}    
    with open(file.filename,encoding='utf-8') as csvf:
        csvReader = csv.DictReader(csvf)
        for rows in csvReader:             
            key = rows['No']
            data[key] = rows    
    return {data}```

Answer 1

下面给出了有关如何将上传的 .csv 文件转换为 JSON 的各种选项。下面的示例中使用了以下 .csv 示例文件。

数据.csv

Id,name,age,height,weight
1,Alice,20,62,120.6
2,Freddie,21,74,190.6
3,Bob,17,68,120.0

选项1

csv.DictReader() 方法也可以接受作为 file 参数的 file objects。 FastAPI 的 UploadFile 使用 Python 的 SpooledTemporaryFile ，一个 file-like 对象(有关更多信息，请查看 this answer )。您可以通过 UploadFile 对象的 .file 属性访问它。但是，由于 FastAPI/Starlette 以 bytes 模式打开文件，如果您将它直接传递给 csv.DictReader() 方法，您会得到一个错误，即 _csv.Error:迭代器应该返回字符串，而不是字节。因此，您可以使用 codecs.iterdecode()(如 this answer 中所建议的那样)，它使用增量解码器迭代解码 iterator 提供的输入(在本例中是从 bytes 到 海峡)。示例:

from fastapi import FastAPI, File, UploadFile
import csv
import codecs

app = FastAPI()
    
@app.post("/upload")
def upload(file: UploadFile = File(...)):
    csvReader = csv.DictReader(codecs.iterdecode(file.file, 'utf-8'))
    data = {}
    for rows in csvReader:             
        key = rows['Id']  # Assuming a column named 'Id' to be the primary key
        data[key] = rows  
    
    file.file.close()
    return data

输出

{
  "1": {
    "Id": "1",
    "name": "Alice",
    "age": "20",
    "height": "62",
    "weight": "120.6"
  },
  "2": {
    "Id": "2",
    "name": "Freddie",
    "age": "21",
    "height": "74",
    "weight": "190.6"
  },
  "3": {
    "Id": "3",
    "name": "Bob",
    "age": "17",
    "height": "68",
    "weight": "120.0"
  }
}

如果您想返回字典的列表，您可以使用下面的方法。由于下面要求 file 在返回结果时 open，因此阻止服务器正确关闭文件(通过调用 file.file.close( )) 完成后，可以使用 BackgroundTasks(在 返回响应后运行)关闭文件:

from fastapi import FastAPI, File, UploadFile, BackgroundTasks
import csv
import codecs

app = FastAPI()

@app.post("/upload")
def upload(background_tasks: BackgroundTasks, file: UploadFile = File(...)):
    csvReader = csv.DictReader(codecs.iterdecode(file.file, 'utf-8'))
    background_tasks.add_task(file.file.close)
    return list(csvReader)

输出

[
  {
    "Id": "1",
    "name": "Alice",
    "age": "20",
    "height": "62",
    "weight": "120.6"
  },
  {
    "Id": "2",
    "name": "Freddie",
    "age": "21",
    "height": "74",
    "weight": "190.6"
  },
  {
    "Id": "3",
    "name": "Bob",
    "age": "17",
    "height": "68",
    "weight": "120.0"
  }
]

选项 2

另一种解决方案是读取上传文件的字节数据——使用contents = file.file.read()(async读/写见this answer) — 然后将字节转换为字符串，最后将它们加载到内存中的文本缓冲区(即 StringIO )，如 here 所述，它可以传递给 csv.DictReader()。示例如下:

from fastapi import FastAPI, File, UploadFile
import csv
from io import StringIO

app = FastAPI()
    
@app.post("/upload")
def upload(file: UploadFile = File(...)):
    data = {}
    contents = file.file.read()
    buffer = StringIO(contents.decode('utf-8'))
    csvReader = csv.DictReader(buffer)
    for row in csvReader:  
        key = row['Id']  # Assuming a column named 'Id' to be the primary key
        data[key] = row  
    
    buffer.close()
    file.file.close()
    return data

选项3

要以您的方式解决问题——即使用文件路径读取 csv 文件，而不是直接使用文件内容或类似文件的对象，如前所述——您可以将文件内容复制到 NamedTemporaryFile 中，与 UploadFile 提供的 SpooledTemporaryFile 不同，它“在文件系统中有一个可见的名称”，“可用于打开文件”(再次检查 this answer 以获得更多信息)。下面是一个工作示例:

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

app = FastAPI()
    
@app.post("/upload")
def upload(file: UploadFile = File(...)):
    data = {}
    temp = NamedTemporaryFile(delete=False)
    try:
        try:
            contents = file.file.read()
            with temp as f:
                f.write(contents);
        except Exception:
            return {"message": "There was an error uploading the file"}
        finally:
            file.file.close()
        
        with open(temp.name,'r', encoding='utf-8') as csvf:
            csvReader = csv.DictReader(csvf)
            for rows in csvReader:             
                key = rows['Id']  # Assuming a column named 'Id' to be the primary key
                data[key] = rows
    except Exception:
        return {"message": "There was an error processing the file"}
    finally:
        #temp.close()  # the `with` statement above takes care of closing the file
        os.remove(temp.name)  # Delete the file
    
    return data

选项4

您还可以将上传文件中的字节写入 BytesIO 流，然后您可以将其转换为 Pandas DataFrame。接下来，使用 to_dict() 方法(如 this answer 中所述)，您可以将数据帧转换为字典并返回它——FastAPI 在幕后使用 jsonable_encoder 将其转换为 JSON 兼容数据，最后，序列化数据并返回 JSONResponse(有关详细信息，请参阅 this answer)。或者，您可以使用 to_json() 方法并直接返回 Response，如选项 1(更新 2)here 中所述。

from fastapi import FastAPI, File, UploadFile
from io import BytesIO
import pandas as pd

app = FastAPI()
    
@app.post("/upload")
def upload(file: UploadFile = File(...)):
    contents = file.file.read()
    buffer = BytesIO(contents)
    df = pd.read_csv(buffer)
    buffer.close()
    file.file.close()
    return df.to_dict(orient='records')

注意:如果文件太大并且占用了所有内存和/或花费太多时间来处理和/或返回结果，请查看 this answer ，以及 this answer 和 this answer.