如何在FastAPI中实现适配自定义格式的通用响应模型

核心思路：利用Pydantic泛型（GenericModel） #

FastAPI基于Pydantic，用GenericModel可以轻松实现通用化的响应结构，无需为每个业务模型重复编写响应模板。我们可以定义两个通用模型：一个处理单个对象，一个处理对象列表，同时支持动态指定数据字段名（比如object/objects）。

代码实现 #

1. 导入依赖 #

from typing import Generic, TypeVar, List, Optional
from pydantic import BaseModel, GenericModel
from fastapi import FastAPI

2. 定义泛型类型变量 #

# 定义泛型类型，代表任意业务模型
T = TypeVar('T')

3. 通用成功响应模型 #

class SingleResponse(GenericModel, Generic[T]):
    status: str = "success"
    data: dict[str, T]

class ListResponse(GenericModel, Generic[T]):
    status: str = "success"
    data: dict[str, List[T]]

4. 业务模型示例 #

class Object(BaseModel):
    id: str
    category: str
    field1: Optional[str] = None

5. 在FastAPI接口中使用 #

app = FastAPI()

# 单个对象接口
@app.get("/object/{obj_id}", response_model=SingleResponse[Object])
def get_object(obj_id: str):
    # 模拟数据库查询
    obj = Object(id=obj_id, category="test", field1="content")
    # 返回指定格式的字典，data里的键对应期望的字段名
    return {"data": {"object": obj}}

# 对象列表接口
@app.get("/objects", response_model=ListResponse[Object])
def get_objects():
    # 模拟列表数据
    objs = [
        Object(id="1", category="test", field1="content"),
        Object(id="2", category="test", field1="content")
    ]
    return {"data": {"objects": objs}}

优化：自动处理单数/复数字段名 #

如果想更自动化，避免手动写object/objects，可以加一个简单的工具函数来生成复数字段名：

def get_plural_name(model_class: type[BaseModel]) -> str:
    # 基础的单数转复数逻辑，可根据业务需求扩展
    class_name = model_class.__name__.lower()
    if class_name.endswith("y"):
        return class_name[:-1] + "ies"
    elif class_name.endswith(("s", "x", "z", "ch", "sh")):
        return class_name + "es"
    else:
        return class_name + "s"

接口中使用示例：

@app.get("/objects", response_model=ListResponse[Object])
def get_objects():
    objs = [Object(id="1", category="test"), Object(id="2", category="test")]
    plural_name = get_plural_name(Object)
    return {"data": {plural_name: objs}}

更简洁的方式：使用装饰器自动包装 #

如果觉得每个接口都写return {"data": {...}}麻烦，可以写一个装饰器自动处理响应包装：

from functools import wraps
import asyncio

def wrap_response(single_field: str = None, list_field: str = None):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            result = await func(*args, **kwargs) if asyncio.iscoroutinefunction(func) else func(*args, **kwargs)
            if isinstance(result, list):
                # 处理列表响应
                field = list_field
                if not field and result:
                    field = get_plural_name(result[0].__class__)
                elif not field:
                    field = "items"
                return {"status": "success", "data": {field: result}}
            else:
                # 处理单个对象响应
                field = single_field or result.__class__.__name__.lower()
                return {"status": "success", "data": {field: result}}
        return wrapper
    return decorator

使用装饰器的接口示例：

@app.get("/object/{obj_id}")
@wrap_response(single_field="object")
def get_object(obj_id: str):
    return Object(id=obj_id, category="test", field1="content")

@app.get("/objects")
@wrap_response(list_field="objects")
def get_objects():
    return [
        Object(id="1", category="test"),
        Object(id="2", category="test")
    ]

这样接口只需返回原始业务对象或列表，装饰器会自动包装成指定格式。

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