下面给出了如何将上传的.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参数。（要了解更多信息，请查看this answer）.您可以通过UploadFile对象的.file属性访问它.但是，由于FastAPI/Starlette以bytes模式打开文件，如果你直接将它传递给csv.DictReader()方法，你会得到一个错误，即_csv.Error: iterator should return strings, not bytes。因此，你可以使用codecs.iterdecode()（如this answer中所建议的），它使用一个增量解码器对 iterator 提供的输入进行迭代解码（在本例中是从bytes到str）。示例：

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"
  }
}

如果您想返回字典的list，您可以使用下面的代码。因为下面的代码要求file在返回结果时为open，从而阻止服务器正确关闭文件（通过调用file.file.close()）当它完成时，可以使用BackgroundTasks（它运行 after 返回一个响应）关闭文件：

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。（如this answer中所述），您可以将 Dataframe 转换为字典并返回它-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.

您之所以会得到Error : FileNotFoundError: [Error 2] No such file or directory : "testdata.csv"，是因为您尝试读取的文件不是本地存储的。
如果要以这种方式读取文件，则应在继续操作之前保存上载的文件：

async def upload(uploaded_file: UploadFile = File(...)):
    # save csv to local dir
    csv_name = uploaded_file.filename
    csv_path = 'path_to/csv_dir/'
    file_path = os.path.join(csv_path, csv_name)
    with open(file_path, mode='wb+') as f:
        f.write(uploaded_file.file.read())
    # read csv and convert to json
    data = {}
    with open(file_path, mode='r', encoding='utf-8') as csvf:
        csvReader = csv.DictReader(csvf)
        for rows in csvReader:             
            key = rows['No']
            data[key] = rows    
    return {data}

异步函数upload()中的file已经打开，可以直接从中取字符，不需要再打开。同样在FastAPI中，类UploadFile实际上是从标准库tempfile.SpooledTemporaryFile派生的，不能通过指定临时文件的路径来访问。
例如，如果使用CPython并在类Unix系统的upload()中读取file.filename的值，它将返回一个数字而不是格式良好的路径，因为类SpooledTemporaryFile的任何示例都将创建文件描述符（在当前存储的数据超过max_size时的某个点），并在访问SpooledTemporaryFile.filename时简单地返回文件描述符（在Unix中应为数字