@提示: 微信搜索【猿码记】回复 【fastapi】即可获取源码信息~

在上篇文章中Python框架篇(2):FastApi-参数接收和验证: https://mp.weixin.qq.com/s/J2_gJxJk2VLfMXgoH1l8Cw
简单学习了一些基础且常用的模型,下面稍微在学点复杂点的模型和使用方法

@注意:对参数parameter目录结构做了优化:

│   ├── types # 声明入参和出参对应的Pydantic模型
│   │   ├── __init__.py
│   │   ├── request # 入参模型
│   │   └── response # 出参模型

1. 参数模型补充

1.1 多参数接收

1.定义模型

app/types/request/demo_param.py文件中,新增内容如下:

...


class StudentParam(BaseModel):
"""
学生信息
"""
name: constr(min_length=2, max_length=4) # 长度
age: conint(ge=18, le=30) # 整数范围:18 <= age <= 30
class_name: str # 班级名称


class ClassInfoParam(BaseModel):
"""
班级信息
"""
class_name: str # 班级名称
class_num: int # 班级人数
teacher_name: str # 老师名称

2.编写路由

app/router下,新增demo_router.py文件,内容如下:

...


@router.post("/query/pydantic/multipleParamReceive")
async def multipleParamReceive(student: request.StudentParam, classInfo: request.ClassInfoParam):
"""
请求体-多参数接收-演示
"""
return {
"msg": "请求体-多参数接收",
"result": {
"student": student,
"classInfo": classInfo,
}
}

3.验证结果

1.2 嵌套模型

1.定义模型

app/types/request/demo_param.py文件中,新增内容如下:

...


class NestedParam(BaseModel):
"""嵌套模型"""
teacher_id: int # 老师id
teacher_name: str # 老师名称
class_list: List[ClassInfoParam] # 老师下班级列表


...

2.编写路由

app/router下,新增demo_router.py文件,内容如下:

...


@router.post("/query/pydantic/nestedModel")
async def nestedModelDemo(param: request.NestedParam):
"""
请求体-嵌套模型接收-演示
"""
return {
"msg": "嵌套模型接收使用-示例",
"result": {
"param": param,
}
}

3.验证结果

2.字段Field

在开发Api和写Api文档的过程中,经常会遇到以下场景:

  1. 把每个字段的中文说明加上,这样方便使用者理解;
  2. 参数设置默认值,如果参数有值则覆盖;
  3. 对于每个字段,最好在文档中,都能给个示例;
  4. 入参名和定义名不一致,如何处理?比如定义的属性是className,入参是class_name;

2.1 参数预览

def Field(
default: Any = Undefined, # 设置参数默认值,场景2
*,
default_factory: Optional[NoArgAnyCallable] = None, # 指定一个函数,该函数的返回值将被用作字段的默认值
alias: Optional[str] = None, # 字段别名,场景4
title: Optional[str] = None,
description: Optional[str] = None, # 字段说明,用于文档生成,,场景1
exclude: Optional[Union['AbstractSetIntStr', 'MappingIntStrAny', Any]] = None,
include: Optional[Union['AbstractSetIntStr', 'MappingIntStrAny', Any]] = None,
const: Optional[bool] = None,
gt: Optional[float] = None, # 条件判断:大于
ge: Optional[float] = None, # 条件判断:等于
lt: Optional[float] = None, # 条件判断:小于
le: Optional[float] = None, # 条件判断:小于等于
multiple_of: Optional[float] = None, # 用于指定数值字段的值必须是某个特定值的倍数
allow_inf_nan: Optional[bool] = None,
max_digits: Optional[int] = None,
decimal_places: Optional[int] = None,
min_items: Optional[int] = None, # 用于验证列表或元组字段的元素个数不少于指定的最小值
max_items: Optional[int] = None, # 用于验证列表或元组字段的元素个数不大于指定的最大值
unique_items: Optional[bool] = None, # 设置true,则验证列表或元组字段的元素不能重复
min_length: Optional[int] = None, # 字符串字段的最小长度
max_length: Optional[int] = None, # 字符串字段的最大长度
allow_mutation: bool = True,
regex: Optional[str] = None, # 正则验证
discriminator: Optional[str] = None,
repr: bool = True,
**extra: Any,
) -> Any:

参数: example,用来给出参数示例

2.2 使用示例

1. 定义模型

app/types/request/demo_param.py文件中,新增内容如下:

class FieldParam(BaseModel):
"""
Field使用示例
"""
name: str = Field(default='', max_length=4, description="填写姓名", example="张三")
age: int = Field(default='', gt=18, description="填写年龄,必须大于18", example=20)
phone: str = Field(default='', description="填写手机号", example="17600000000", regex=r'^1\d{10}$')
likes: List[str] = Field(default='[]', description="填写爱好", example=["篮球", "足球"], min_items=2,
unique_items=True)

2.查看文档

3.响应模型

为了演示方便,上面示例都是直接返回一个字典dict
,来代替响应模型,正常情况下,会封装个通用模型,并通过方法调用,好处有以下几点:

  • 保持输出结构整体风格统一;
  • 新增通用字段,不用在每个接口都赋值;比如时间戳等;
  • 对响应内容统一加日志,这样就不用在每个接口都加上一遍;
  • 针对错误会有些报警之类的处理,在统一地方加即可;

3.1 定义结构

{
"code": 200,
"msg": "处理成功",
"data": ...,
"additional": {
"time": "2023-12-04 19:00:23",
"trace_id": "cc1b12a5dfee26a7dcc29fe47dcfbde0"
}
}

3.2 定义模型

新增文件app/types/response/http_resp.py,内容如下:

from datetime import datetime
from typing import Any
from pydantic import BaseModel, Field
from app.utils import StringUtil


# ---------------------- 定义模型 ----------------------
class Additional(BaseModel):
"""额外信息"""
time: str
trace_id: str


class HttpResponse(BaseModel):
"""http统一响应"""
code: int = Field(default=200) # 响应码
msg: str = Field(default="处理成功") # 响应信息
data: Any | None # 具体数据
additional: Additional # 额外信息

3.3 响应方法

app/types/response/http_resp.py文件中,新增内容如下:

def ResponseSuccess(resp: Any) -> HttpResponse:
"""成功响应"""
currentTime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return HttpResponse(
data=resp,
additional=Additional(
time=currentTime,
trace_id=StringUtil.GenerateMd5(currentTime),
))


def ResponseFail(msg: str, code: int = -1) -> HttpResponse:
"""响应失败"""
currentTime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return HttpResponse(
code=code,
msg=msg,
additional=Additional(
time=currentTime,
trace_id=StringUtil.GenerateMd5(currentTime),
))

3.4 使用示例

1. 定义接口

app/router下,新增demo_router.py文件,内容如下:

# 导入前,需要先在app/types/response/__init__.py,引入http_resp
from app.types import response


@router.post("/resp/demo", summary="响应模型示例")
async def respDemo(param: request.FieldParam) -> response.HttpResponse:
"""
响应模型示例-演示
"""
if "游戏" in param.likes:
return response.ResponseFail("禁止玩游戏~")

return response.ResponseSuccess(param)

2.文档调用

4. 在线生成模型

当我们有了json后,可以直接通过这个网站:https://jsontopydantic.com/进行生成,省的我们挨个去写