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

问题描述 投票：0回答：3

我正在尝试将我的 

.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}```
python csv upload fastapi csvtojson
3个回答
19
投票

下面提供了有关如何将上传的 

.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
参数文件对象。 FastAPI 的 
UploadFile
 使用 Python 的 
SpooledTemporaryFile
，一个类似文件的对象（有关更多信息，请查看这个答案）。您可以通过 .file
 对象的
UploadFile
 属性访问它。但是，由于 FastAPI/Starlette 以
bytes
 模式打开文件，如果直接将其传递给
csv.DictReader()
 方法，则会收到错误，即
_csv.Error: iterator should return strings, not bytes
。因此，您可以使用
codecs.iterdecode()
（如这个答案中所建议的），它使用增量解码器来迭代解码迭代器提供的输入（在本例中从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
 （运行返回响应后）关闭文件：
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
 读/写，请参阅 
这个答案
），然后将字节转换为字符串，最后将它们加载到内存中文本缓冲区（即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 
，与 
SpooledTemporaryFile 提供的 
UploadFile
 不同，“在文件系统中有一个可见的名称”，“可用于打开文件”（再次，请查看 
这个答案
 以获取更多信息） 。下面是一个工作示例：
from fastapi import FastAPI, File, UploadFile, HTTPException
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:
            raise HTTPException(status_code=500, detail='Something went wrong')
        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:
        raise HTTPException(status_code=500, detail='Something went wrong when 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()
 方法（如
这个答案中所述），您可以将数据帧转换为字典并返回它，FastAPI 在幕后将使用以下方法转换为 JSON 兼容的数据： jsonable_encoder
，最后，序列化数据并返回
JSONResponse
（有关更多详细信息，请参阅
这个答案）。作为更快的替代方案，您可以使用 to_json()
 方法并直接返回自定义 
Response，如
此答案
的选项 1（更新 2）中所述。
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')



注意
：如果文件太大并占用了所有内存和/或花费太多时间来处理和/或返回结果，请查看此答案以及这个答案和这个答案。
    




    

    
    
    

    

        

            

                

                    
0
投票

                    
                

            

        

        
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}


    



    

    
    
    

    

        

            

                

                    
0
投票

                    
                

            

        

        
file
中的
upload()
已经打开，可以直接从中获取字符，无需再次打开。同样在FastAPI中，类
UploadFile
实际上是从标准库
tempfile.SpooledTemporaryFile
派生的，无法通过指定临时文件的路径来访问。

例如，如果您使用 CPython 并在类 Unix 系统中的 
file.filename
 中读取 
upload()
 的值，它会返回一个数字而不是格式良好的路径，因为类 
SpooledTemporaryFile
 的任何实例都会创建一个文件描述符（当当前存储的数据超过
max_size
时）并在访问时简单地返回文件描述符（应该是Unix中的数字）
SpooledTemporaryFile.filename
    

	

    

    


  

  

    
最新问题


  





  

    © www.soinside.com 2019 - 2024. All rights reserved.