基于Pydantic和FastAPI动态处理未知Schema的多CSV文件

针对你遇到的动态CSV Schema适配、类型推断、响应模型动态化及大数据性能问题，以下是具体实现方案：

1. 运行时动态创建Pydantic模型 #

利用Pydantic提供的create_model函数，结合CSV文件的表头和字段类型，可在运行时生成对应模型。核心思路是先解析CSV的字段信息，再构建字段字典传入create_model。

示例代码：

from pydantic import create_model, Field, Optional
import pandas as pd
from typing import List

def build_dynamic_model(csv_path: str, model_name: str) -> type:
    # 读取CSV前100行推断字段类型（行数可根据数据调整）
    df = pd.read_csv(csv_path, nrows=100)
    field_definitions = {}
    
    # 遍历列映射Pydantic类型
    for col in df.columns:
        dtype = df[col].dtype
        # 基础类型映射
        if pd.api.types.is_integer_dtype(dtype):
            field_type = int
        elif pd.api.types.is_float_dtype(dtype):
            field_type = float
        elif pd.api.types.is_datetime64_dtype(dtype):
            field_type = pd.Timestamp
        else:
            field_type = str
        
        # 空值处理：若列存在空值，设为Optional类型
        if df[col].isnull().any():
            field_type = Optional[field_type]
        
        # 定义字段（可选字段默认值为None）
        field_definitions[col] = (field_type, Field(default=None))
    
    # 强制ID字段为必填整数（已知ID类型）
    if "id" in field_definitions:
        field_definitions["id"] = (int, ...)
    
    # 生成动态模型
    return create_model(model_name, **field_definitions)

2. 高效推断字段类型与可选字段处理 #

• 类型推断：借助pandas的read_csv自动推断字段类型，比手动解析更高效；若需更精准控制，可扩展类型判断逻辑（比如识别布尔值、枚举值）。
• 可选字段处理：通过检查列是否存在空值，自动将字段标记为Optional，并设置默认值None；对于必填字段（如ID），直接强制覆盖为必填类型（(int, ...)表示无默认值，必须传值）。

优化点：若CSV存在大量重复Schema的情况，可缓存已生成的模型（见性能部分），避免重复推断。

3. 动态适配ResponseModel #

同样使用create_model生成动态响应模型，将每个CSV对应的动态模型作为字段类型。

示例代码：

def build_response_model(csv_model_map: dict) -> type:
    # csv_model_map格式：{"fruits": FruitDynamicModel, "animals": AnimalDynamicModel, ...}
    response_fields = {}
    for key, model in csv_model_map.items():
        # 每个字段为可选的模型列表，默认值为None
        response_fields[key] = (Optional[List[model]], None)
    
    return create_model("DynamicResponseModel", **response_fields)

使用时，先为每个CSV生成动态模型，再传入build_response_model得到最终的响应模型：

csv_paths = {
    "fruits": "./data/fruits.csv",
    "animals": "./data/animals.csv",
    "cities": "./data/cities.csv",
    "books": "./data/books.csv",
    "cars": "./data/cars.csv",
    "planets": "./data/planets.csv"
}

# 生成所有CSV对应的动态模型
dynamic_models = {name: build_dynamic_model(path, name.capitalize()) for name, path in csv_paths.items()}

# 生成动态响应模型
DynamicResponse = build_response_model(dynamic_models)

4. 60MB数据处理的性能注意事项 #

（1）缓存动态模型 #

避免每次请求都重新生成模型，可基于CSV文件的修改时间或哈希值缓存模型：

from functools import lru_cache
import os

@lru_cache(maxsize=6)  # 最多缓存6个模型（对应6个CSV）
def get_cached_model(csv_path: str, model_name: str, file_mtime: float) -> type:
    return build_dynamic_model(csv_path, model_name)

# 使用时传入文件修改时间，确保模型与最新CSV同步
for name, path in csv_paths.items():
    mtime = os.path.getmtime(path)
    dynamic_models[name] = get_cached_model(path, name.capitalize(), mtime)

（2）分块读取CSV #

一次性加载60MB数据会占用大量内存，使用pandas的chunksize分块处理：

def process_csv_chunks(csv_path: str, model: type) -> List:
    validated_data = []
    # 每1000行处理一次，可根据内存调整chunk大小
    for chunk in pd.read_csv(csv_path, chunksize=1000):
        # 转换为字典列表后批量验证
        chunk_records = chunk.to_dict("records")
        validated_chunk = [model.model_validate(record) for record in chunk_records]
        validated_data.extend(validated_chunk)
    return validated_data

（3）使用StreamingResponse返回数据 #

避免一次性生成完整JSON占用内存，用流式响应逐步输出：

from fastapi import FastAPI, StreamingResponse
import json

app = FastAPI()

@app.get("/api/data")
async def get_csv_data():
    def generate_response():
        yield "{"
        first_key = True
        for name, path in csv_paths.items():
            if not first_key:
                yield ", "
            first_key = False
            # 获取缓存的动态模型
            mtime = os.path.getmtime(path)
            model = get_cached_model(path, name.capitalize(), mtime)
            
            yield f'"{name}": ['
            first_row = True
            for chunk in pd.read_csv(path, chunksize=1000):
                for record in chunk.to_dict("records"):
                    if not first_row:
                        yield ", "
                    first_row = False
                    # 验证后转为JSON字符串输出
                    validated = model.model_validate(record)
                    yield json.dumps(validated.model_dump())
            yield "]"
        yield "}"
    
    return StreamingResponse(generate_response(), media_type="application/json")

（4）启用Pydantic的性能优化 #

• 使用Pydantic v2+版本，其底层基于pydantic-core，验证速度比v1快数倍。
• 批量验证时，可使用model_validate_batch（Pydantic v2新增）替代循环验证，进一步提升效率：

  validated_chunk = model.model_validate_batch(chunk_records)
  

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