FastAPI高级实践：高效上传文件与复杂JSON数据（含列表和字典）

from typing import List, Optional
from pydantic import BaseModel, Field
from fastapi import Query

class QueryParams(BaseModel):
    width: Optional[float] = Field(None)
    height: Optional[float] = Field(None)
    words: List[str] = Field(Query(...)) # 明确声明为列表查询参数

from fastapi import FastAPI, status, Form, UploadFile, File, Depends, Query, HTTPException
from pydantic import BaseModel, Field, ValidationError
from fastapi.encoders import jsonable_encoder
from typing import Optional, List
import json

app = FastAPI()

# 定义查询参数模型
class BaseParams(BaseModel):
    width: Optional[float] = Field(None)
    height: Optional[float] = Field(None)
    words: List[str] = Field(Query(...)) # 列表型查询参数

# 定义嵌套的JSON对象模型
class BaseBox(BaseModel):
    l: float = Field(...)
    t: float = Field(...)
    r: float = Field(...)
    b: float = Field(...)

# 定义复杂的JSON数据模型
class Base(BaseModel):
    boxes: List[BaseBox] = Field(...)
    comments: List[str] = Field(...)
    code: int = Field(...)

# 自定义依赖项，用于解析作为表单字符串传输的JSON数据
def parse_json_form_data(data: str = Form(...)):
    try:
        # 尝试将字符串解析为Base模型
        return Base.model_validate_json(data)
    except ValidationError as e:
        # 如果解析失败，抛出422错误
        raise HTTPException(
            detail=jsonable_encoder(e.errors()),
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        )

@app.post("/submit")
def submit(
    base_params: BaseParams = Depends(),      # 依赖注入查询参数
    base: Base = Depends(parse_json_form_data), # 依赖注入解析后的JSON数据
    files: List[UploadFile] = File(...),      # 接收文件列表
):
    """
    接收查询参数、JSON数据（作为表单字符串）和文件列表。
    """
    return {
        "Params": base_params,
        "JSON Payload": base,
        "Filenames": [file.filename for file in files],
    }

# 启动应用：uvicorn app:app --reload

curl -X 'POST' \
  'http://localhost:8000/submit?width=10.5&height=20.0&words=hello&words=world' \
  -H 'accept: application/json' \
  -H 'Content-Type: multipart/form-data' \
  -F 'data={"boxes": [{"l": 0,"t": 0,"r": 0,"b": 0}, {"l": 10,"t": 10,"r": 20,"b": 20}], "comments": ["foo", "bar"], "code": 123}' \
  -F 'files=@./test_image.png;type=image/png' \
  -F 'files=@./another_file.txt;type=text/plain'

from fastapi import FastAPI, Body, UploadFile, File, Depends, Query, HTTPException
from pydantic import BaseModel, Field, model_validator, ValidationError
from typing import Optional, List
import json

app = FastAPI()

# 定义查询参数模型
class BaseParams(BaseModel):
    width: Optional[float] = Field(None)
    height: Optional[float] = Field(None)
    words: List[str] = Field(Query(...))

# 定义嵌套的JSON对象模型
class BaseBox(BaseModel):
    l: float = Field(...)
    t: float = Field(...)
    r: float = Field(...)
    b: float = Field(...)

# 定义复杂的JSON数据模型，并添加model_validator
class Base(BaseModel):
    boxes: List[BaseBox] = Field(...)
    comments: List[str] = Field(...)
    code: int = Field(...)

    # Pydantic v2的model_validator，在模型实例化前对值进行预处理
    @model_validator(mode="before")
    @classmethod
    def validate_to_json(cls, value):
        if isinstance(value, str):
            try:
                return cls(**json.loads(value))
            except json.JSONDecodeError as e:
                raise ValueError(f"Invalid JSON string for Base model: {e}")
        return value

@app.post("/submit")
def submit(
    base_params: BaseParams = Depends(), # 依赖注入查询参数
    base: Base = Body(...),              # Pydantic模型作为请求体，由model_validator处理
    files: List[UploadFile] = File(...), # 接收文件列表
):
    """
    接收查询参数、JSON数据（由model_validator处理）和文件列表。
    """
    return {
        "Params": base_params,
        "JSON Payload": base,
        "Filenames": [file.filename for file in files],
    }

# 启动应用：uvicorn app:app --reload

curl -X 'POST' \
  'http://localhost:8000/submit?width=10.5&height=20.0&words=alpha&words=beta' \
  -H 'accept: application/json' \
  -H 'Content-Type: multipart/form-data' \
  -F 'base={"boxes": [{"l": 0,"t": 0,"r": 0,"b": 0}], "comments": ["hello", "world"], "code": 456}' \
  -F 'files=@./document.pdf;type=application/pdf'