如何在FastAPI POST请求中同时传文件与JSON？CSV列选择方法

你这段代码没法正常运行的核心原因是：HTTP POST请求只能有一种Content-Type，而你同时要求FastAPI接收application/json格式的DataConfiguration请求体，以及multipart/form-data格式的文件上传，这两种格式无法在同一个请求里共存。

下面给你两种可行的解决方案，同时演示如何从上传的CSV文件中提取你需要的列：

方案一：将JSON配置作为表单字段传递 #

这种方法把JSON配置打包成字符串，作为表单的一个字段和文件一起上传，是最接近你原有需求的方式：

from typing import List
from pydantic import BaseModel
from fastapi import FastAPI, UploadFile, File, Form
import pandas as pd

app = FastAPI()

class DataConfiguration(BaseModel):
    textColumnNames: List[str]
    idColumn: str

@app.post("/data")
async def data(
    config: str = Form(...),  # 接收JSON格式的配置字符串
    csvFile: UploadFile = File(...)
):
    # 把表单里的JSON字符串解析成DataConfiguration对象
    data_config = DataConfiguration.parse_raw(config)
    
    # 用pandas读取上传的CSV文件（如果不用pandas，也可以用标准库csv模块）
    df = pd.read_csv(csvFile.file)
    
    # 整理需要提取的列名
    target_columns = [data_config.idColumn] + data_config.textColumnNames
    
    # 校验CSV里是否存在这些列
    missing_columns = [col for col in target_columns if col not in df.columns]
    if missing_columns:
        return {"error": f"CSV文件中缺少以下必填列: {missing_columns}"}
    
    # 提取指定列的数据
    selected_data = df[target_columns].to_dict(orient="records")
    
    return {
        "status": "success",
        "selected_columns": target_columns,
        "data": selected_data
    }

调用这个接口时，你需要用multipart/form-data格式发送请求：

• 一个字段叫config，值是符合DataConfiguration结构的JSON字符串（比如{"textColumnNames": ["content", "title"], "idColumn": "user_id"}）
• 另一个字段叫csvFile，值是要上传的CSV文件

方案二：将配置拆分为查询参数（适合简单场景） #

如果你的配置项不多，也可以直接把参数放到URL的查询字符串里，这样代码会更简洁：

from typing import List
from fastapi import FastAPI, UploadFile, File, Query
import pandas as pd

app = FastAPI()

@app.post("/data")
async def data(
    textColumnNames: List[str] = Query(...),  # 多个文本列可以重复传，比如?textColumnNames=content&textColumnNames=title
    idColumn: str = Query(...),
    csvFile: UploadFile = File(...)
):
    df = pd.read_csv(csvFile.file)
    target_columns = [idColumn] + textColumnNames
    
    # 同样做列存在性校验
    missing_columns = [col for col in target_columns if col not in df.columns]
    if missing_columns:
        return {"error": f"CSV文件中缺少以下必填列: {missing_columns}"}
    
    selected_data = df[target_columns].to_dict(orient="records")
    
    return {
        "status": "success",
        "selected_columns": target_columns,
        "data": selected_data
    }

调用这个接口时，直接在URL里拼接查询参数，比如：
POST /data?idColumn=user_id&textColumnNames=content&textColumnNames=title
同时上传CSV文件即可。

不用pandas的替代方案（用标准库） #

如果你不想依赖pandas，也可以用Python标准库的csv模块来处理文件：

import csv
# 替换方案一里的pandas读取部分
csv_file = csvFile.file
csv_file.seek(0)  # 确保文件指针回到开头
reader = csv.DictReader(csv_file)

selected_data = []
for row in reader:
    # 只保留需要的列
    filtered_row = {col: row[col] for col in target_columns}
    selected_data.append(filtered_row)

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