参数—请求体"/>
【Python开发】FastAPI 03:请求参数—请求体
除了路径参数和查询参数,还有请求体,其用于传递 JSON、XML 或其他格式的数据,以便服务器能够读取并做出相应的处理,可以说请求体的作用更为强大。试想一下,如果存在七八个参数,路径参数和查询是不是就招架不住了,但是请求体则可以将这七八个参数一网打尽。
目录
1 声明请求体
1.1 使用 Pydantic 模型
① 创建数据模型
② 声明为参数
③ 请求体 + 路径参数 + 查询参数
④ 数据模型使用 Field
1.2 使用 Body
① 单一值的请求体
② 嵌入单个请求体参数
2 请求体中的嵌套模型
2.1 List 字段
① 未声明元素类型
② 声明元素类型
2.2 Set 类型
2.3 嵌套模型
2.4 特殊的类型和校验
① HttpUrl
② 纯列表请求体
3 请求体—更新数据
3.1 JSON 兼容编码器
3.2 更新数据
① 用 PUT 更新数据
② 用 PATCH 进行部分更新
③ 小结
4 额外数据类型及参数
4.1 其他数据类型
4.2 Cookie 参数
4.3 Header 参数
① 声明 Header 参数
② 自动转换
📌源码地址:
1 声明请求体
当需要将数据从客户端(例如浏览器)发送给 API(可以理解为是应用后台)时,你将其作为「请求体」发送。
请求体是客户端发送给 API 的数据,响应体是 API 发送给客户端的数据。
你不能使用 GET 操作(HTTP 方法)发送请求体。
要发送请求体,必须使用下列方法之一:POST(较常见)、PUT、DELETE 或 PATCH。
1.1 使用 Pydantic 模型
① 创建数据模型
# 导入 Pydantic 的 BaseModel
from pydantic import BaseModel#将数据模型声明为继承自 BaseModel 的类,使用标准的 Python 类型来声明所有属性
class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = None
类似查询参数,某字段若非必须,可为其设置一个默认值或可选 None,否则该字段就是必须的。
上面的模型声明了一个这样的 JSON「object」(或 Python dict):
{"name": "Foo","description": "An optional description","price": 45.2,"tax": 3.5
}
因为 description 和 tax 是可选的(它们的默认值为 None),那么下面的 JSON「object」也是有效的:
{"name": "Foo","price": 45.2
}
② 声明为参数
类似路径参数和查询参数,可以使用相同的方式声明请求体,并且将它的类型声明为你创建的 Item 模型,完整代码:
from typing import Union
from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Noneapp = FastAPI()@app.post("/items1/")
async def create_item(item: Item):return item
此时,我们可以使用交互文档、postman、apifox等工具进行请求测试,后边将使用交互文档进行测试。
仅仅使用了 Python 类型声明,FastAPI 将做以下事情:
- 将请求体作为 JSON 读取。
- 校验数据。
- 如果数据无效,将返回一条清晰易读的错误信息,指出不正确数据的确切位置和内容。
- 将接收的数据赋值到参数 item 中。
- 这些模式将成为生成的 OpenAPI 模式的一部分,并且被自动化文档 UI 所使用。
在函数内部,你可以直接访问模型对象的所有属性:
...@app.post("/items2/")
async def create_item(item: Item):item_dict = item.dict()if item.tax:price_with_tax = item.price + item.taxitem_dict.update({"price_with_tax": price_with_tax})return item_dict
📌 文档
你所定义模型的 JSON 模式将可以在交互式 API 文档(http://127.0.0.1:8000/docs#)中展示:
不仅如此,你也可以在该文档中访问该接口:
然后就可以看到响应啦:
③ 请求体 + 路径参数 + 查询参数
你还可以同时声明请求体、路径参数和查询参数。
...@app.put("/items3/{item_id}")
async def create_item(item_id: int, item: Item, q: Union[str, None] = None):result = {"item_id": item_id, **item.dict()}if q:result.update({"q": q})return result
请求参数将依次按如下规则进行识别:
- 如果在路径中也声明了该参数,它将被用作路径参数。
- 如果参数属于单一类型(比如 int、float、str、bool 等)它将被解释为查询参数。
- 如果参数的类型被声明为一个 Pydantic 模型,它将被解释为请求体。
④ 数据模型使用 Field
与使用 Query、Path 类似,你可以使用 Pydantic 的 Field 在 Pydantic 模型内部声明校验和元数据。
from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel, Field #导入 Fieldapp = FastAPI()class Item1(BaseModel): #对模型属性使用 Fieldname: strdescription: Union[str, None] = Field(default=None, title="The description of the item", max_length=300)price: float = Field(gt=0, description="The price must be greater than zero")tax: Union[float, None] = None@app.put("/items4/{item_id}")
async def update_item(item_id: int, item: Item1):results = {"item_id": item_id, "item": item}return results
Field 的工作方式和 Query 和 Path 相同,包括它们的参数等等也完全相同,详情可以查看上一章内容~
1.2 使用 Body
与使用 Query 和 Path 为查询参数和路径参数定义额外数据的方式相同,FastAPI 提供了一个同等的 Body。
① 单一值的请求体
如果该请求参数是一个单一值(比如 str、int 等),但是依旧想让他以请求体的形式存在,那么可以使用 Body 来修饰,不然系统将默认他是一个查询参数。
from typing import Unionfrom fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()class User(BaseModel):username: strfull_name: Union[str, None] = None@app.put("/items5/{item_id}")
async def update_item(item_id: int, user: User, single: int = Body()):results = {"item_id": item_id, "user": user, "single": single}return results
此时,FastAPI 期望的请求体如下:
{"user": {"username": "yinyu","full_name": "yinyuyu"},"single": 5
}
② 嵌入单个请求体参数
如果你想要这个 JSON 外再带一个 key,那么可以使用嵌入的特殊 Body 参数:
user: User = Body(embed=True)
比如:
...@app.put("/items6/{item_id}")
async def update_item(item_id: int, user: User = Body(embed=True)):results = {"item_id": item_id, "user": user}return results
此时,FastAPI 期望的请求体如下:
{"user": {"username": "yinyu","full_name": "yinyuyu"}
}
而不是:
{"username": "yinyu","full_name": "yinyuyu"
}
2 请求体中的嵌套模型
使用 FastAPI,你可以定义、校验、记录文档并使用任意深度嵌套的模型(归功于 Pydantic)
2.1 List 字段
① 未声明元素类型
你可以将一个请求体模型的属性/字段定义为拥有子元素的类型。例如 Python list:
from typing import List, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Model1(BaseModel):name: strprice: floattags: list = []@app.put("/model1/{model_id}")
async def update_model(model_id: int, model: Model1):results = {"model_id": model_id, "model": model}return results
此时,FastAPI 期望的请求体可以如下:
{"name": "string","price": 0,"tags": [1, "yinyu"]
}
因为 tags 只是一个由元素组成的列表,它没有声明每个元素的类型,因此可以放多个类型的元素进去。
② 声明元素类型
如果我们想指定列表中的元素类型,那么 Python 也提供了一种特定的方法:
要声明具有子类型的类型,例如 list、dict、tuple:
- 从 typing 模块导入它们
- 使用方括号 [ ] 将子类型作为「类型参数」传入
from typing import Listmy_list: List[str]
因此,在我们的示例中,我们可以将 tags 明确地指定为一个「字符串列表」:
...class Model2(BaseModel):name: strprice: floattags: List[str] = [] #声明具有子类型的列表,要声明具有子类型的类型,例如 list、dict、tuple:@app.put("/model2/{model_id}")
async def update_model(model_id: int, model: Model2):results = {"model_id": model_id, "model": model}return results
2.2 Set 类型
如果你希望列表中的元素不能重复,那么 set 这一数据类型可以很好的满足该要求:
from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Model3(BaseModel):name: strprice: floatmtags: Set[str] = set()@app.put("/model3/{model_id}")
async def update_model(model_id: int, model: Model3):results = {"model_id": model_id, "model": model}return results
此时,FastAPI 期望的请求体可以如下:
{"name": "yinyu","price": 0,"mtags": ["handsome","handsome"]
}
正确请求后的响应内容则如下:
{"model_id": 1,"model": {"name": "string","price": 0,"mtags": ["1"]}
}
这样,即使你收到带有重复数据的请求,后台也会将重复元素进行筛选后保留一项。
2.3 嵌套模型
Pydantic 模型的每个属性/字段都具有类型,那么你可以使得该属性也是一个 Pydantic 模型
from typing import Union
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()#定义一个 Image 模型作为子模型
class Image(BaseModel):url: strname: strclass Base1(BaseModel):name: strdescription: Union[str, None] = Noneimage: Union[Image, None] = None #将子模型用作类型@app.put("/base1/{base_id}")
async def update_base(base_id: int, base: Base1):results = {"item_id": base_id, "item": base}return results
此时,FastAPI 期望的请求体可以如下:
{"name": "yinyu","description": "The pretender","image": {"url": "","name": "The yinyu live"}
}
2.4 特殊的类型和校验
① HttpUrl
在 Image 模型中我们有一个 url 字段,我们可以把它声明为 Pydantic 的 HttpUrl,而不是 str,类似的特殊类型可以在 Field Types - Pydantic 页面查看。
from pydantic import BaseModel, HttpUrlclass Image(BaseModel):url: HttpUrlname: str
② 纯列表请求体
如果你期望的 JSON 请求体的最外层是一个 JSON array(即 Python list),则可以在路径操作函数的参数中声明此类型,就像声明 Pydantic 模型一样:
from typing import Listfrom fastapi import FastAPI
from pydantic import BaseModel, HttpUrlapp = FastAPI()class Image(BaseModel):url: HttpUrlname: str@app.post("/images/multiple/")
async def create_multiple_images(images: List[Image]):return images
此时,FastAPI 期望的请求体可以如下 👇,可以有 n 个 Image。
[{"url": "www.11","name": "yin"},{"url": "www.22","name": "yu"}
]
3 请求体—更新数据
3.1 JSON 兼容编码器
在某些情况下,您可能需要将数据类型(如 Pydantic 模型)转换为与 JSON 兼容的数据类型(如dict、lis t等)。 对于这种要求, FastAPI提供了jsonable_encoder()函数。
假设你有一个数据库名为 fake_db,它只能接收与 JSON 兼容的数据。 例如,它不接收 datetime 这类的对象,因为这些对象与 JSON 不兼容。 因此,datetime 对象必须将转换为包含 ISO 格式化的 str 类型对象。 同样,这个数据库也不会接收 Pydantic 模型(带有属性的对象),而只接收 dict。 对此你可以使用 jsonable_encoder 。 它
接收一个对象,比如 Pydantic 模型,并会返回一个 JSON 兼容的版本:
from datetime import datetime
from typing import List, Unionfrom fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModelapp = FastAPI()fake_db = {}class Item(BaseModel):title: strtimestamp: datetimedescription: Union[str, None] = None@app.put("/items/{id}")
def update_item(id: str, item: Item):json_compatible_item_data = jsonable_encoder(item)fake_db[id] = json_compatible_item_data
在这个例子中,它将Pydantic模型转换为 dict,并将 datetime 转换为 str。 调用它的结果后就可以使用 Python 标准编码中的 json.dumps()。
3.2 更新数据
① 用 PUT 更新数据
更新数据请用 HTTP PUT 操作。 把输入数据转换为以 JSON 格式存储的数据(比如,使用 NoSQL 数据库时),可以使用 jsonable_encoder。例如,把 datetime 转换为 str。
from typing import List, Union
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: Union[str, None] = Nonedescription: Union[str, None] = Noneprice: Union[float, None] = Nonetax: float = 10.5tags: List[str] = []items = {"foo": {"name": "Foo", "price": 50.2},"bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},"baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}@app.put("/items2/{item_id}") #PUT 用于接收替换现有数据的数据。
async def update_item(item_id: str, item: Item):update_item_encoded = jsonable_encoder(item)items[item_id] = update_item_encodedreturn items
② 用 PATCH 进行部分更新
HTTP PATCH 操作用于更新部分数据。 即只发送要更新的数据,其余数据保持不变。
...@app.patch("/items3/{item_id}")
async def update_item(item_id: str, item: Item):stored_item_data = items[item_id]stored_item_model = Item(**stored_item_data)update_data = item.dict(exclude_unset=True)updated_item = stored_item_model.copy(update=update_data)items[item_id] = jsonable_encoder(updated_item)return updated_item
📌 使用 Pydantic 的 exclude_unset 参数
update_data = item.dict(exclude_unset=True)
更新部分数据时,可以在 Pydantic 模型的 .dict() 中使用 exclude_unset 参数。
比如,item.dict(exclude_unset=True)。 这段代码生成的 dict 只包含创建 item 模型时显式设置的数据,而不包括默认值。
📌 使用 Pydantic 的 update 参数
接下来,用 .copy() 为已有模型创建调用 update 参数的副本,该参数为包含更新数据的 dict。
例如:
updated_item = stored_item_model.copy(update=update_data)
③ 小结
简而言之,更新部分数据应:
- 使用 PATCH 而不是 PUT (可选,也可以用 PUT);
- 提取存储的数据;
- 把数据放入 Pydantic 模型;
- 生成不含输入模型默认值的 dict (使用 exclude_unset 参数);
- 只更新用户设置过的值,不用模型中的默认值覆盖已存储过的值。
- 为已存储的模型创建副本,用接收的数据更新其属性 (使用 update 参数)。
- 把模型副本转换为可存入数据库的形式(比如,使用 jsonable_encoder)。
- 这种方式与 Pydantic 模型的 .dict() 方法类似,但能确保把值转换为适配 JSON 的数据类型,例如, 把 datetime 转换为 str 。 把
- 数据保存至数据库;
- 返回更新后的模型。
实际上,HTTP PUT 也可以完成相同的操作。 但本节以 PATCH 为例的原因是,该操作就是为了这种用例创建的。
4 额外数据类型及参数
4.1 其他数据类型
Python 同样支持更复杂的数据类型:
- UUID:
- 一种标准的 "通用唯一标识符" ,在许多数据库和系统中用作ID。
- 在请求和响应中将以 str 表示。
- datetime.datetime:
- 一个 Python datetime.datetime.
- 在请求和响应中将表示为 ISO 8601 格式的 str ,比如: 2008-09-15T15:53:00+05:00.
- datetime.date:
- Python datetime.date.
- 在请求和响应中将表示为 ISO 8601 格式的 str ,比如: 2008-09-15.
- datetime.time:
- 一个 Python datetime.time.
- 在请求和响应中将表示为 ISO 8601 格式的 str ,比如: 14:23:55.003.
- datetime.timedelta:
- 一个 Python datetime.timedelta.
- 在请求和响应中将表示为 float 代表总秒数。
- Pydantic 也允许将其表示为 "ISO 8601 时间差异编码", 查看文档了解更多信息。
- frozenset:
- 在请求和响应中,作为 set 对待:
- 在请求中,列表将被读取,消除重复,并将其转换为一个 set。
- 在响应中 set 将被转换为 list 。
- 产生的模式将指定那些 set 的值是唯一的 (使用 JSON 模式的 uniqueItems)。
- bytes:
- 标准的 Python bytes。
- 在请求和相应中被当作 str 处理。
- 生成的模式将指定这个 str 是 binary "格式"。
- Decimal:
- 标准的 Python Decimal。
- 在请求和相应中被当做 float 一样处理。
比如下面的例子,就使用了上述的一些类型:
@app.put("/items/{item_id}")
async def read_items(item_id: UUID,start_datetime: Union[datetime, None] = Body(default=None),end_datetime: Union[datetime, None] = Body(default=None),repeat_at: Union[time, None] = Body(default=None),process_after: Union[timedelta, None] = Body(default=None),
):start_process = start_datetime + process_afterduration = end_datetime - start_processreturn {"item_id": item_id,"start_datetime": start_datetime,"end_datetime": end_datetime,"repeat_at": repeat_at,"process_after": process_after,"start_process": start_process,"duration": duration,}
其中函数内的参数有原生的数据类型,可以执行正常的日期操作。
4.2 Cookie 参数
声明 Cookie 参数的结构与声明 Query 参数和 Path 参数时相同,第一个值是参数的默认值,同时也可以传递所有验证参数或注释参数,来对 Cookie 参数进行校验:
from typing import Unionfrom fastapi import Cookie, FastAPIapp = FastAPI()@app.get("/items1/")
async def read_items(ads_id: Union[str, None] = Cookie(default=None)):return {"ads_id": ads_id}
必须使用 Cookie 来声明 cookie 参数,否则参数将会被解释为查询参数。
4.3 Header 参数
你可以使用定义 Query、Path 和 Cookie 参数一样的方法定义 Header 参数。
① 声明 Header 参数
同样,第一个值是默认值,你可以传递所有的额外验证或注释参数:
from typing import Unionfrom fastapi import FastAPI, Headerapp = FastAPI()@app.get("/items2/")
async def read_items(user_agent: Union[str, None] = Header(default=None)):return {"User-Agent": user_agent}
必须使用 Header 来声明 Header 参数,否则参数将会被解释为查询参数。
② 自动转换
大多数标准的 headers (请求头)用 "连字符" 分隔,比如 user-agent,但是像这样的变量在 Python 中是无效的。因此, 默认情况下, Header 将把参数名称的字符从下划线 (_) 转换为连字符 (-) 来提取并记录 headers。
同时,HTTP headers 是大小写不敏感的,你可以像通常在 Python 代码中那样使用 user_agent ,而不需要将首字母大写为 User_Agent 或类似的东西。
如果出于某些原因,你需要禁用下划线到连字符的自动转换,那么可设置 Header 的参数 convert_underscores 为 False:
from typing import Unionfrom fastapi import FastAPI, Headerapp = FastAPI()@app.get("/items/")
async def read_items(strange_header: Union[str, None] = Header(default=None, convert_underscores=False)
):return {"strange_header": strange_header}
更多推荐
【Python开发】FastAPI 03:请求参数—请求体
发布评论