如何通过REST API传输多文件并读取指定CSV合并为Pandas数据帧

下面分步骤实现你的需求：

1. 处理REST API的文件传输 #

这里分两种常见场景，分别给出实现示例：

场景1：用FastAPI接收客户端上传的多文件 #

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

app = FastAPI()

@app.post("/process-files/")
async def process_uploaded_files(files: list[UploadFile] = File(...)):
    # 后续筛选、读取、合并逻辑写在此处

场景2：从外部API批量下载目标文件 #

import requests
from io import StringIO

# 替换为实际的文件下载URL列表
file_urls = [
    "https://example.com/file1_data.csv",
    "https://example.com/file2_data.csv",
    "https://example.com/file1_prices.csv",
    "https://example.com/file2_prices.csv"
]

# 批量下载文件并保存内容与文件名
downloaded_files = []
for url in file_urls:
    resp = requests.get(url)
    resp.raise_for_status()  # 确保请求成功
    downloaded_files.append({
        "filename": url.split("/")[-1],
        "content": resp.text
    })

2. 筛选并读取指定CSV文件 #

不管是上传还是下载的文件，都按文件名标识筛选，分别读取为DataFrame后合并同类型文件：

处理上传文件的逻辑（放在FastAPI接口函数内） #

data_frames = []
prices_frames = []

for file in files:
    filename = file.filename
    # 将二进制文件内容转为字符串流，供pandas读取
    csv_content = StringIO((await file.read()).decode("utf-8"))
    
    if "_data.csv" in filename:
        data_frames.append(pd.read_csv(csv_content))
    elif "_prices.csv" in filename:
        prices_frames.append(pd.read_csv(csv_content))

# 合并所有_data类型文件为一个DataFrame
df_combined_data = pd.concat(data_frames, ignore_index=True)
# 合并所有_prices类型文件为一个DataFrame
df_combined_prices = pd.concat(prices_frames, ignore_index=True)

处理下载文件的逻辑 #

data_frames = []
prices_frames = []

for file in downloaded_files:
    filename = file["filename"]
    csv_content = StringIO(file["content"])
    
    if "_data.csv" in filename:
        data_frames.append(pd.read_csv(csv_content))
    elif "_prices.csv" in filename:
        prices_frames.append(pd.read_csv(csv_content))

df_combined_data = pd.concat(data_frames, ignore_index=True)
df_combined_prices = pd.concat(prices_frames, ignore_index=True)

3. 合并两个类型的DataFrame #

合并时需指定关联键（如product_id、date等两张表共有的列），根据业务需求选择合并方式：

# 示例：以product_id为关联键，内连接合并（仅保留两边都匹配的数据）
final_merged_df = pd.merge(
    df_combined_data,
    df_combined_prices,
    on="product_id",  # 替换为你的实际关联列名
    how="inner"
)

# 若需保留所有数据，改用outer连接：
# final_merged_df = pd.merge(df_combined_data, df_combined_prices, on="product_id", how="outer")

# 查看合并结果
print(final_merged_df.head())

关键注意事项 #

• 确保同类型CSV文件的列结构一致，否则concat会出现列不匹配问题，可提前校验：

  if data_frames:
      base_cols = data_frames[0].columns
      for df in data_frames[1:]:
          assert set(df.columns) == set(base_cols), "所有_data.csv文件列结构必须一致"
  

• 合并前确认关联键的正确性，避免出现笛卡尔积（数据行数异常暴增）。
• 处理大文件时，用分块读取避免内存溢出：

  # 分块读取示例
  chunk_size = 10000
  data_chunks = []
  for file in downloaded_files:
      if "_data.csv" in file["filename"]:
          for chunk in pd.read_csv(StringIO(file["content"]), chunksize=chunk_size):
              data_chunks.append(chunk)
  df_combined_data = pd.concat(data_chunks, ignore_index=True)
  

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