书接上文,多个关联模型这种情况很常见。

特别是用户模型,因为:

  • 输入模型应该含密码
  • 输出模型不应含密码
  • 数据库模型需要加密的密码

千万不要存储用户的明文密码。始终存储可以进行验证的安全哈希值

如果不了解这方面的知识,请参阅安全性中的章节,了解什么是密码哈希

多个模型

下面的代码展示了不同模型处理密码字段的方式,及使用位置的大致思路:

Python 3.10+ Python 3.8+

  1. from fastapi import FastAPI
  2. from pydantic import BaseModel, EmailStr
  3. app = FastAPI()
  4. class UserIn(BaseModel):
  5. username: str
  6. password: str email: EmailStr
  7. full_name: str | None = None
  8. class UserOut(BaseModel):
  9. username: str
  10. email: EmailStr
  11. full_name: str | None = None
  12. class UserInDB(BaseModel):
  13. username: str
  14. hashed_password: str email: EmailStr
  15. full_name: str | None = None
  16. def fake_password_hasher(raw_password: str):
  17. return "supersecret" + raw_password
  18. def fake_save_user(user_in: UserIn):
  19. hashed_password = fake_password_hasher(user_in.password) user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password) print("User saved! ..not really")
  20. return user_in_db
  21. @app.post("/user/", response_model=UserOut) async def create_user(user_in: UserIn):
  22. user_saved = fake_save_user(user_in)
  23. return user_saved
  1. from typing import Union
  2. from fastapi import FastAPI
  3. from pydantic import BaseModel, EmailStr
  4. app = FastAPI()
  5. class UserIn(BaseModel):
  6. username: str
  7. password: str email: EmailStr
  8. full_name: Union[str, None] = None
  9. class UserOut(BaseModel):
  10. username: str
  11. email: EmailStr
  12. full_name: Union[str, None] = None
  13. class UserInDB(BaseModel):
  14. username: str
  15. hashed_password: str email: EmailStr
  16. full_name: Union[str, None] = None
  17. def fake_password_hasher(raw_password: str):
  18. return "supersecret" + raw_password
  19. def fake_save_user(user_in: UserIn):
  20. hashed_password = fake_password_hasher(user_in.password) user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password) print("User saved! ..not really")
  21. return user_in_db
  22. @app.post("/user/", response_model=UserOut) async def create_user(user_in: UserIn):
  23. user_saved = fake_save_user(user_in)
  24. return user_saved

**user_in.dict() 简介

Pydantic 的 .dict()

user_in 是类 UserIn 的 Pydantic 模型。

Pydantic 模型支持 .dict() 方法,能返回包含模型数据的字典

因此,如果使用如下方式创建 Pydantic 对象 user_in

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

就能以如下方式调用:

user_dict = user_in.dict()

现在,变量 user_dict中的就是包含数据的字典(变量 user_dict 是字典,不是 Pydantic 模型对象)。

以如下方式调用:

print(user_dict)

输出的就是 Python 字典

{ 'username': 'john', 'password': 'secret', 'email': 'john.doe@example.com', 'full_name': None, }

解包 dict

字典 user_dict**user_dict 形式传递给函数(或类),Python 会执行解包操作。它会把 user_dict 的键和值作为关键字参数直接传递。

因此,接着上面的 user_dict 继续编写如下代码:

UserInDB(**user_dict)

就会生成如下结果:

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

或更精准,直接把可能会用到的内容与 user_dict 一起使用:

UserInDB( username = user_dict["username"], password = user_dict["password"], email = user_dict["email"], full_name = user_dict["full_name"], )

用其它模型中的内容生成 Pydantic 模型

上例中 ,从 user_in.dict() 中得到了 user_dict,下面的代码:

user_dict = user_in.dict() UserInDB(**user_dict)

等效于:

UserInDB(**user_in.dict())

……因为 user_in.dict() 是字典,在传递给 UserInDB 时,把 ** 加在 user_in.dict() 前,可以让 Python 进行解包

这样,就可以用其它 Pydantic 模型中的数据生成 Pydantic 模型。

解包 dict 和更多关键字

接下来,继续添加关键字参数 hashed_password=hashed_password,例如:

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

……输出结果如下:

UserInDB( username = user_dict["username"], password = user_dict["password"], email = user_dict["email"], full_name = user_dict["full_name"], hashed_password = hashed_password, )

警告

辅助的附加函数只是为了演示可能的数据流,但它们显然不能提供任何真正的安全机制。

减少重复

FastAPI 的核心思想就是减少代码重复。

代码重复会导致 bug、安全问题、代码失步等问题(更新了某个位置的代码,但没有同步更新其它位置的代码)。

上面的这些模型共享了大量数据,拥有重复的属性名和类型。

FastAPI 可以做得更好。

声明 UserBase 模型作为其它模型的基类。然后,用该类衍生出继承其属性(类型声明、验证等)的子类。

所有数据转换、校验、文档等功能仍将正常运行。

这样,就可以仅声明模型之间的差异部分(具有明文的 password、具有 hashed_password 以及不包括密码)。

通过这种方式,可以只声明模型之间的区别(分别包含明文密码、哈希密码,以及无密码的模型)。

Python 3.10+Python 3.8+

  1. from fastapi import FastAPI
  2. from pydantic import BaseModel, EmailStr
  3. app = FastAPI()
  4. class UserBase(BaseModel):
  5. username: str
  6. email: EmailStr
  7. full_name: str | None = None
  8. class UserIn(UserBase):
  9. password: str
  10. class UserOut(UserBase):
  11. pass
  12. class UserInDB(UserBase):
  13. hashed_password: str
  14. def fake_password_hasher(raw_password: str):
  15. return "supersecret" + raw_password
  16. def fake_save_user(user_in: UserIn):
  17. hashed_password = fake_password_hasher(user_in.password)
  18. user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
  19. print("User saved! ..not really")
  20. return user_in_db
  21. @app.post("/user/", response_model=UserOut)
  22. async def create_user(user_in: UserIn):
  23. user_saved = fake_save_user(user_in)
  24. return user_saved
  1. from typing import Union
  2. from fastapi import FastAPI
  3. from pydantic import BaseModel, EmailStr
  4. app = FastAPI()
  5. class UserBase(BaseModel):
  6. username: str
  7. email: EmailStr
  8. full_name: Union[str, None] = None
  9. class UserIn(UserBase):
  10. password: str
  11. class UserOut(UserBase):
  12. pass
  13. class UserInDB(UserBase):
  14. hashed_password: str
  15. def fake_password_hasher(raw_password: str):
  16. return "supersecret" + raw_password
  17. def fake_save_user(user_in: UserIn):
  18. hashed_password = fake_password_hasher(user_in.password)
  19. user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
  20. print("User saved! ..not really")
  21. return user_in_db
  22. @app.post("/user/", response_model=UserOut)
  23. async def create_user(user_in: UserIn):
  24. user_saved = fake_save_user(user_in)
  25. return user_saved

Union 或者 anyOf

响应可以声明为两种类型的 Union 类型,即该响应可以是两种类型中的任意类型。

在 OpenAPI 中可以使用 anyOf 定义。

为此,请使用 Python 标准类型提示 typing.Union

笔记

定义 Union 类型时,要把详细的类型写在前面,然后是不太详细的类型。下例中,更详细的 PlaneItem 位于 Union[PlaneItem,CarItem] 中的 CarItem 之前。

Python 3.10+Python 3.8+

  1. from typing import Union
  2. from fastapi import FastAPI
  3. from pydantic import BaseModel
  4. app = FastAPI()
  5. class BaseItem(BaseModel):
  6. description: str
  7. type: str
  8. class CarItem(BaseItem):
  9. type: str = "car"
  10. class PlaneItem(BaseItem):
  11. type: str = "plane" size: int
  12. items = {
  13. "item1": {"description": "All my friends drive a low rider", "type": "car"},
  14. "item2": {
  15. "description": "Music is my aeroplane, it's my aeroplane",
  16. "type": "plane",
  17. "size": 5,
  18. },
  19. }
  20. @app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem]) async def read_item(item_id: str):
  21. return items[item_id]
  1. from typing import Union
  2. from fastapi import FastAPI
  3. from pydantic import BaseModel
  4. app = FastAPI()
  5. class BaseItem(BaseModel):
  6. description: str
  7. type: str
  8. class CarItem(BaseItem):
  9. type: str = "car"
  10. class PlaneItem(BaseItem):
  11. type: str = "plane" size: int
  12. items = {
  13. "item1": {"description": "All my friends drive a low rider", "type": "car"},
  14. "item2": {
  15. "description": "Music is my aeroplane, it's my aeroplane",
  16. "type": "plane",
  17. "size": 5,
  18. },
  19. }
  20. @app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem]) async def read_item(item_id: str):
  21. return items[item_id]

模型列表

使用同样的方式也可以声明由对象列表构成的响应。

为此,请使用标准的 Python typing.List

Python 3.9+Python 3.8+

  1. from fastapi import FastAPI
  2. from pydantic import BaseModel
  3. app = FastAPI()
  4. class Item(BaseModel):
  5. name: str
  6. description: str
  7. items = [
  8. {"name": "Foo", "description": "There comes my hero"},
  9. {"name": "Red", "description": "It's my aeroplane"},
  10. ]
  11. @app.get("/items/", response_model=list[Item]) async def read_items():
  12. return items
  1. from typing import List
  2. from fastapi import FastAPI
  3. from pydantic import BaseModel
  4. app = FastAPI()
  5. class Item(BaseModel):
  6. name: str
  7. description: str
  8. items = [
  9. {"name": "Foo", "description": "There comes my hero"},
  10. {"name": "Red", "description": "It's my aeroplane"},
  11. ]
  12. @app.get("/items/", response_model=List[Item]) async def read_items():
  13. return items

任意 dict 构成的响应

任意的 dict 都能用于声明响应,只要声明键和值的类型,无需使用 Pydantic 模型。

事先不知道可用的字段 / 属性名时(Pydantic 模型必须知道字段是什么),这种方式特别有用。

此时,可以使用 typing.Dict

Python 3.9+Python 3.8+

  1. from fastapi import FastAPI
  2. app = FastAPI()
  3. @app.get("/keyword-weights/", response_model=dict[str, float]) async def read_keyword_weights():
  4. return {"foo": 2.3, "bar": 3.4}
  1. from typing import Dict
  2. from fastapi import FastAPI
  3. app = FastAPI()
  4. @app.get("/keyword-weights/", response_model=Dict[str, float]) async def read_keyword_weights():
  5. return {"foo": 2.3, "bar": 3.4}

小结

针对不同场景,可以随意使用不同的 Pydantic 模型继承定义的基类。

实体必须具有不同的状态时,不必为不同状态的实体单独定义数据模型。例如,用户实体就有包含 password、包含 password_hash 以及不含密码等多种状态。