敲代码不忘补水

敲代码不忘补水

关注
发信
关注(28)粉丝(399)

FastAPI 响应模型详解：单一模型、多模型组合、基类继承复用、联合模型 (Union)等多场景应用

FastAPI 响应模型详解：单一模型、多模型组合、基类继承复用、联合模型 (Union)等多场景应用

本篇文章详细讲解了如何在 FastAPI 中定义和管理响应模型，包括单一模型、多模型组合、基类继承复用、联合模型 (Union)、列表 (List) 和字典 (Dict) 的响应结构。通过精细的代码示例，读者将学习如何处理请求与响应数据不同时的情况、优化字段显示，以及动态选择不同的数据模型以适应复杂需求。此外，文章还重点介绍了减少代码重复的方法，例如通过基类继承定义共性字段，确保代码简洁高效。这篇文章适合从初学者到高级开发者，帮助您全面掌握 FastAPI 的响应模型，构建功能完善、结构清晰的高质量 API。

在 API 开发中，定义清晰的响应数据结构是提升接口可靠性和用户体验的关键。FastAPI 通过 Pydantic 模型提供了强大的响应模型功能，支持多种场景下的灵活应用。以下示例中使用的 Python 版本为 Python 3.10.15，FastAPI 版本为 0.115.4。

一 多种响应模型

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from typing import Union

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str | None = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None


class UserInDB(BaseModel):
    username: str
    hashed_password: str
    email: EmailStr
    full_name: str | None = None


def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password


def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db


@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in)
    return user_saved

在实际应用中，常见的情况是需要多个相关的模型，特别是在处理用户数据时。例如：

  • 输入模型：应包含明文密码，用于用户注册。
  • 输出模型：不应包含密码，以保护用户隐私。
  • 数据库模型：应存储密码的加密哈希值，而非明文。

运行代码文件 chapter16.py 来启动应用：

$ uvicorn chapter16:app --reload

在 SwaggerUI 中可以查看在线文档：http://127.0.0.1:8000/docs 。

划重点
1 Pydantic 的 **user_in.dict() 方法详解

Pydantic 模型支持 .dict() 方法，用于将模型的数据转换为 Python 字典。例如，如果创建了一个 UserIn 类的 Pydantic 对象 user_in：

user_in = UserIn(
  username="john", 
  password="secret", 
  email="john.doe@example.com")

可以通过以下方式调用 .dict() 方法，将其转换为字典：

user_dict = user_in.dict()

此时，user_dict 就是包含模型数据的字典，而不再是 Pydantic 模型对象。例如，调用：

print(user_dict)

输出结果将是：

{
    'username': 'john',
    'password': 'secret',
    'email': 'john.doe@example.com',
    'full_name': None,
}
2 字典的解包操作

使用 **user_dict 可以将字典的键值对作为关键字参数传递给函数或类。例如，可以用解包的方式将字典传递给另一个 Pydantic 模型：

UserInDB(**user_dict)

这相当于直接将字典的每个键值作为参数传递，等效于以下代码：

UserInDB(
    username="john",
    password="secret",
    email="john.doe@example.com",
    full_name=None,
)
3 使用 .dict() 生成其他 Pydantic 模型

通过 user_in.dict() 可以直接创建包含数据的字典，然后将其传递给另一个 Pydantic 模型：

UserInDB(**user_in.dict())

这与之前的步骤等效，因为 .dict() 方法生成的字典被解包传递给新的模型。

4 添加额外字段参数

在创建新模型的同时，还可以添加额外的参数。例如，可以为 UserInDB 模型添加一个 hashed_password：

UserInDB(**user_in.dict(), hashed_password=hashed_password)

生成的对象如下：

UserInDB(
    username="john",
    password="secret",
    email="john.doe@example.com",
    full_name=None,
    hashed_password=hashed_password,
)

通过这种方式，可以灵活地在原有数据基础上添加或修改字段，生成新的 Pydantic 模型实例。

二 继承复用响应模型

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None


class UserIn01(UserBase):
    password: str


class UserOut01(UserBase):
    pass


class UserInDB01(UserBase):
    hashed_password: str


def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password


def fake_save_user(user_in: UserIn01):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB01(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db


@app.post("/user01/", response_model=UserOut01)
async def create_user(user_in: UserIn01):
    user_saved = fake_save_user(user_in)
    return user_saved

FastAPI 的核心之一是减少代码重复，以避免错误、安全隐患和更新不一致的问题。为了解决上述模型之间的重复，可以声明一个 UserBase 基类，并派生其他模型继承其属性和验证逻辑，只需定义各模型之间的差异部分（如明文密码、哈希密码或不含密码）。这样既减少了重复，又确保数据转换、验证和文档生成正常运行。

三 Union 多种响应模型（弱水三千只取一瓢）

class BaseItem(BaseModel):
    description: str
    type: str


class CarItem(BaseItem):
    type: str = "car"


class PlaneItem(BaseItem):
    type: str = "plane"
    size: int


class PlaneItem01(BaseItem):
    type: str = "plane"
    size: int


items = {
    "item1": {"description": "All my friends drive a low rider", "type": "car"},
    "item2": {
        "description": "Music is my aeroplane, it's my aeroplane",
        "type": "plane",
        "size": 5,
    },
    "item02": {
        "description": "Music is my aeroplane, it's my aeroplane",
        "type": "planes",
        "size": 5,
    },
}


@app.get("/items/{item_id}", response_model=Union[PlaneItem, PlaneItem01, CarItem])
async def read_item(item_id: str):
    return items[item_id]

可以使用 Python 的类型 typing.Union 来定义多种类型的响应，即响应可以声明为 Union 类型，响应数据可以是多种类型中的任意一种。

四 typing.List 响应模型

class Item(BaseModel):
    name: str
    description: str


items = [
    {"name": "Foo", "description": "There comes my hero"},
    {"name": "Red", "description": "It's my aeroplane"},
]


@app.get("/items/", response_model=list[Item])
async def read_items():
    return items

五 typing.Dict 响应模型

@app.get("/keyword-weights/", response_model=dict[str, float])
async def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}

任意的字典都可用于声明响应，只需声明键和值的类型，无需使用 Pydantic 模型。在未知字段名时，可以使用 typing.Dict。

六 完整代码示例

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from typing import Union

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str | None = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None


class UserInDB(BaseModel):
    username: str
    hashed_password: str
    email: EmailStr
    full_name: str | None = None


def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password


def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db


@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in)
    return user_saved


# 继承复用

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None


class UserIn01(UserBase):
    password: str


class UserOut01(UserBase):
    pass


class UserInDB01(UserBase):
    hashed_password: str


def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password


def fake_save_user(user_in: UserIn01):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB01(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db


@app.post("/user01/", response_model=UserOut01)
async def create_user(user_in: UserIn01):
    user_saved = fake_save_user(user_in)
    return user_saved


class BaseItem(BaseModel):
    description: str
    type: str


class CarItem(BaseItem):
    type: str = "car"


class PlaneItem(BaseItem):
    type: str = "plane"
    size: int


class PlaneItem01(BaseItem):
    type: str = "plane"
    size: int


items = {
    "item1": {"description": "All my friends drive a low rider", "type": "car"},
    "item2": {
        "description": "Music is my aeroplane, it's my aeroplane",
        "type": "plane",
        "size": 5,
    },
    "item02": {
        "description": "Music is my aeroplane, it's my aeroplane",
        "type": "planes",
        "size": 5,
    },
}


@app.get("/items/{item_id}", response_model=Union[PlaneItem, PlaneItem01, CarItem])
async def read_item(item_id: str):
    return items[item_id]


class Item(BaseModel):
    name: str
    description: str


items = [
    {"name": "Foo", "description": "There comes my hero"},
    {"name": "Red", "description": "It's my aeroplane"},
]


@app.get("/items/", response_model=list[Item])
async def read_items():
    return items


@app.get("/keyword-weights/", response_model=dict[str, float])
async def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}

七 源码地址

详情见：GitHub FastApiProj

八 参考

[1] FastAPI 文档

12-08 08:27
Powered by LMLPHP ©2024 敲代码不忘补水 0.004955