本文参考FastAPI教程https://fastapi.tiangolo.com/zh/tutorial
第一步
import uvicorn
from fastapi import FastAPIapp = FastAPI()@app.get("/")
async def root():return {"message": "Hello World"}if __name__ == '__main__':uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)
$ uvicorn main:app --reload
- 导入
FastAPI
:FastAPI
是一个为你的API提供了所有功能的Python类 app = FastAPI()
创建一个FastAPI
实例,这个实例将是创建你所有API的主要交互对象。这个app
同样在如下命令中被uvicorn
所引用。- 创建一个路径操作。
【路径】:这里的【路径】指的是从URL中第一个/
起的后半部分,比如在https://example.com/items/foo
中,路径是/items/foo
,【路径】也通常被称为【端点】或【路由】。
【操作】:这里的【操作】是指一种HTTP【方法】,如POST
——创建数据,GET
——读取数据,PUT
——更新数据,DELETE
——删除数据,以及更少见的几种OPTIONS
,HEAD
,PATCH
,TRACE
,在HTTP协议中,你可以使用以上的其中一种(或多种)【方法】与每个路径进行通信。 - 定义一个路径操作装饰器:
@app.get("/")
告诉FastAPI
在它下方的函数负责处理如下访问请求:请求路径为/
,使用get
操作。(@something
语法在Python中被称为【装饰器】,接收位于其下方的函数并且用它完成一些工作,这里是路径操作装饰器。) - 定义路径操作函数:位于路径操作装饰器下的函数,这里是
async def root()
,这个例子用的是async
函数,其作用在后面讲。 - 返回内容:可以返回一个
dict
、list
,像str
、int
一样的值,等等。还可以返回Pydanic
模型(后面会说),还有许多其他将会自动转换为JSON的对象和模型(包括ORM对象等)。
查看运行结果
- 打开浏览器访问 http://127.0.0.1:8000。讲看到如下的JSON相应:
{"message": "Hello World"}
交互式API文档
- 跳转到 http://127.0.0.1:8000/docs。你将会看到自动生成的交互式API文档(由
Swagger UI
提供)。
可选的API文档
- 前往 http://127.0.0.1:8000/redoc。你将会看到可选的自动生成文档(由
ReDoc
提供)。
路径参数
FastAPI支持以下路径模板语法定义动态路由,声明路径参数(变量):
import uvicorn
from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}/{go}")
async def read_item(go):return {"item_id": go}if __name__ == '__main__':uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)
这段代码声明了相应位置上的路径参数item_id
和go
,并把go
的值传递给路径函数的参数go
。
运行示例并访问http://127.0.0.1:8000/items/para1/para2
可获得如下相应:
{"item_id":"para2"}
声明路径参数的类型
使用Python标准类型注释,声明路径操作函数中路径参数的类型。
from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(item_id: int):return {"item_id": item_id}
上例把item_id
的类型声明为int
。
检查:类型声明将为函数提供错误检查、代码补全等编辑器支持
数据转换
运行上述声明item_id
的类型为int
的例子并访问 http://127.0.0.1:8000/items/3
,返回的相应如下:
{"item_id":3}
可见,函数接收并返回的值是3
(int
),不是"3"
(str
)。FastAPI通过类型声明自动解析请求中的数据。
数据校验
如果通过浏览器访问http://127.0.0.1:8000/items/foo
,将会接收如下HTTP错误信息:
{"detail": [{"type": "int_parsing","loc": ["path","item_id"],"msg": "Input should be a valid integer, unable to parse string as an integer","input": "foo"}]
}
因为路径参数item_id
的值("foo"
)的类型不是int
。
检查:FastAPI使用Python类型声明实现了数据校验,上面的错误清晰的指出了未通过校验的具体原因,这在开发调试与API交互的代码时非常有用。
查看文档
访问 http://127.0.0.1:8000/docs,查看自动生成的 API 文档:
Pydantic
FastAPI可以充分利用Pydanic的优势,用它在后台校验数据。
路径操作的顺序
from fastapi import FastAPIapp = FastAPI()@app.get("/users/me")
async def read_user_me():return {"user_id": "the current user"}@app.get("/users/{user_id}")
async def read_user(user_id: str):return {"user_id": user_id}
/users/me
和/users/{user_id}
不能反过来,否则/users/me
的路径会被/users/{user_id}
接收。
预设值
路径操作使用Python的Enum
类型接收预设的路径参数。
from enum import Enumfrom fastapi import FastAPIclass ModelName(str, Enum):alexnet = "alexnet"resnet = "resnet"lenet = "lenet"app = FastAPI()@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):if model_name is ModelName.alexnet:return {"model_name": model_name, "message": "Deep Learning FTW!"}if model_name.value == "lenet":return {"model_name": model_name, "message": "LeCNN all the images"}return {"model_name": model_name, "message": "Have some residuals"}
- 导入
Enum
并创建继承自str
和Enum
的子类。通过从str
继承,API文档就能把值的类型定义为字符串,并且能正确渲染。然后创建包含固定值的类属性,这些固定值是可用的有效值。 - 使用
Enum
类(ModelName
)创建使用类型注解的路径参数。 - API文档会显示预定义路径参数的可用值。
- 使用Python枚举类型:路径参数的值是枚举的元素。枚举类
ModelName
中的枚举元素支持比较操作(if model_name is ModelName.alexnet
),使用model_name.value
获取枚举值。 - 返回枚举元素:即使嵌套在JSON请求体里(例如,
dict
),也可以从路径操作返回枚举元素。
包含路径的路径参数
from fastapi import FastAPIapp = FastAPI()@app.get("/files/{file_path:path}")
async def read_file(file_path: str):return {"file_path": file_path}
参数名为file_path
,结尾部分的:path
说明该参数应匹配路径。
查询参数
声明的参数不是路径参数时,路径操作函数会把该参数自动解释为查询参数。
from fastapi import FastAPIapp = FastAPI()fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):return fake_items_db[skip : skip + limit]
查询字符串是键值对的集合,这些键值对位于URL的?
之后,以&
分隔。
例如,以下URL中:
http://127.0.0.1:8000/items/?skip=0&limit=2
查询参数为:skip: 0
,limit: 2
这些值都是URL的组成部分,因此它们的类型本应是字符串。
但声明Python类型(上例中为int
)之后,这些值就会转换为声明的类型,并进行类型校验。
所有应用于路径参数的流程也适用于查询参数。
默认值
查询参数由默认值,如上例。
可以进行如下URL访问:
- http://127.0.0.1:8000/items/
- http://127.0.0.1:8000/items/?skip=0&limit=10
- http://127.0.0.1:8000/items/?skip=20
可选参数
把默认值设为None
即可声明可选的查询参数:
from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None):if q:return {"item_id": item_id, "q": q}return {"item_id": item_id}
检查:注意,FastAPI 可以识别出 item_id
是路径参数,q
不是路径参数,而是查询参数。
查询参数类型转换
FastAPI会自动转换参数类型:
from typing import Unionfrom fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Union[str, None] = None, short: bool = False):item = {"item_id": item_id}if q:item.update({"q": q})if not short:item.update({"description": "This is an amazing item that has a long description"})return item
在本例中访问:
http://127.0.0.1:8000/items/foo?short=1
或
http://127.0.0.1:8000/items/foo?short=True
或short=on,short=true,short=yes或其它任意大小写形式,函数接收到的short
参数都是布尔值True
。False
同理(0,false,off,no)。
多个路径和查询参数
FastAPI可以识别同时声明的多个路径参数和查询参数,而且声明查询参数的顺序并不重要,FastAPI通过参数名进行检测:
from typing import Unionfrom fastapi import FastAPIapp = FastAPI()@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: str, q: Union[str, None] = None, short: bool = False
):item = {"item_id": item_id, "owner_id": user_id}if q:item.update({"q": q})if not short:item.update({"description": "This is an amazing item that has a long description"})return item
把不是路径参数的参数(至此只有查询参数)声明为默认值,或者是吧默认值设为None
,这样参数就不是必选的,否则是必选的。
上述代码可以用以下URL进行测试:
http://127.0.0.1:8000/users/2/items/item_id?q=test&short=0
将会得到这样的相应:
{"item_id":"item_id","owner_id":2,"q":"test","description":"This is an amazing item that has a long description"}
再比如:
from typing import Unionfrom fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str, skip: int = 0, limit: Union[int, None] = None
):item = {"item_id": item_id, "needy": needy, "skip": skip, "limit": limit}return item
上例中有3个查询参数:needy
,必选的str
类型参数,skip
,默认值为0
的int
类型参数,limit
,可选的int
类型参数。
请求体
FastAPI使用请求体从客户端(例如浏览器)向API发送数据。
请求体是客户端发送给API的数据。响应体是API发送给客户端的数据。
API基本上肯定要发送响应体,但是客户端不一定发送请求体。
使用Pydantic模型声明请求体,能充分利用它的功能和优点。
使用Pydantic声明请求体
from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneapp = FastAPI()@app.post("/items/")
async def create_item(item: Item):return item
步骤如下:
- 从
pydantic
中导入BaseModel
- 创建数据类型:把数据模型声明为继承
BaseModel
的类。使用Python标准类型声明所有属性。 - 声明请求体参数:使用与声明路径和查询参数相同的方式声明请求体,把请求体添加至路径操作,
@app.post()
及其下面的函数async def create_item(item: Item):...
,此处请求体参数的类型为Item
类型。
使用如下代码进行测试:
import requests# 定义请求的 JSON 数据
item_data = {"name": "Item1","description": "This is item 1","price": 19.99,"tax": 2.00
}# 发送 POST 请求
response = requests.post("http://localhost:8000/items/", json=item_data)# 打印返回的 JSON 数据
print(response.json())
可以看到,FastAPI接收来自测试代码的请求,返回:
{'name': 'Item1', 'description': 'This is item 1', 'price': 19.99, 'tax': 2.0}
结论
仅使用Python类型声明,FastAPI就可以
- 以JSON形式读取请求体
- (在必要时)把请求体转换为对应的类型
- 校验数据:数据无效时返回错误信息,并指出错误数据的确切位置和内容
- 把接收的数据赋值给参数
item
- 为模型生成JSON Schema,在项目中所需的位置使用
使用模型
在路径操作函数内部直接访问模型对象的属性:
from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneapp = FastAPI()@app.post("/items/")
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
用测试代码测试结果如下:
{'description': 'Mechanical keyboard','name': 'Keyboard','price': 49.99,'price_with_tax': 54.99,'tax': 5.0}
请求体+路径参数
FastAPI支持同时声明路径参数和请求体。
FastAPI能识别与路径参数匹配的函数参数,还能识别从请求体中获取的类型为Pydantic模型的函数参数。
from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneapp = FastAPI()@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):return {"item_id": item_id, **item.dict()}
使用如下测试代码发送PUT
请求:
import requests
from pprint import pprint# 定义要发送的 JSON 数据,符合 Item 模型的定义
item_data = {"name": "Mouse","description": "Wireless mouse","price": 19.99,"tax": 1.50
}# 定义要更新的 item_id
item_id = 1# 发送 PUT 请求
response = requests.put(f"http://localhost:8000/items/{item_id}", json=item_data)# 打印返回的 JSON 数据
pprint(response.json())
返回结果为:
{'description': 'Wireless mouse','item_id': 1,'name': 'Mouse','price': 19.99,'tax': 1.5}
请求体+路径参数+查询参数
FastAPI支持同时声明请求体、路径参数和查询参数。
from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneapp = FastAPI()@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: str | None = None):result = {"item_id": item_id, **item.dict()}if q:result.update({"q": q})return result
函数参数按如下规则进行识别:
- 路径中声明了相同参数的参数,是路径参数
- 类型是(
int
,float
,str
,bool
等)单类型的参数,是查询参数 - 类型是Pydantic模型的参数,是请求体
使用如下测试代码:
import requests
from pprint import pprint# 定义要发送的 JSON 数据,符合 Item 模型的定义
item_data = {"name": "Keyboard","description": "Mechanical keyboard","price": 49.99,"tax": 5.00
}# 定义要更新的 item_id
item_id = 1# 定义查询参数 q
q_param = "example"# 发送 PUT 请求
response = requests.put(f"http://localhost:8000/items/{item_id}?q={q_param}", json=item_data)# 打印返回的 JSON 数据
pprint(response.json())
可得到返回结果:
{'description': 'Mechanical keyboard','item_id': 1,'name': 'Keyboard','price': 49.99,'q': 'example','tax': 5.0}
查询参数和字符串校验
FastAPI允许你为参数声明额外的信息和校验。
以下面的应用程序为例:
from fastapi import FastAPIapp = FastAPI()@app.get("/items/")
async def read_items(q: str | None = None):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results
查询参数q
的类型为str
,默认值为None
,因此它是可选的。
额外的校验
我们打算添加约束条件:即使q
是可选的,但只要提供了该参数,则该参数值不能超过50个字符的长度
具体步骤如下:
- 导入
Query
:首先从fastapi
导入Query
。 - 使用
Query
作为默认值:将Query
用作查询参数的默认值,并将它的max_length
参数设置为50,由于我们必须用Query(default=None)
替换默认值None
,Query
的第一个参数同样是用于定义默认值。
from typing import Unionfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, max_length=50)):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results
添加更多的校验
- 还可以添加
min_length
参数 - 还可以添加正则表达式:定义一个参数值必须匹配的正则表达式
from typing import Unionfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, min_length=3, max_length=50, pattern="^fixedquery$"),
):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results
声明为必需参数
当我们不需要声明额外的校验或元数据时,只需不声明默认值就可以使q
参数成为必需参数,如:
q: str
但是我们现在正在用Query
声明它,如:
q: Union[str, None] = Query(default=None, min_length=3)
因此,当你在使用Query
且需要声明一个值是必需的时,只需不声明默认参数:
from fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: str = Query(min_length=3)):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results
显式声明必需参数的方法
- 使用
(...)
声明:q: str = Query(default=..., min_length=3)
- 可以声明
None
为一个有效的类型,仍是必需参数:q: Union[str, None] = Query(default=..., min_length=3)
- 使用Pydantic中的
Required
代替省略号...
:from pydantic import Required
,q: str = Query(default=Required, min_length=3)
大多数情况,隐式省略default
参数就够了,通常不必使用显式声明...
或Required
查询参数列表/多个值
当你使用Query
显式地定义查询参数时,你还可以声明它去接收一组值,或换句话来说,接收多个值。
例如,要声明一个可在URL中出现多次的查询参数q
,可以这样写:
from typing import List, Unionfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Union[List[str], None] = Query(default=None)):query_items = {"q": q}return query_items
然后输入以下网址:
http://localhost:8000/items/?q=foo&q=bar
你会在路径操作函数的函数参数q
中以一个Pythonlist
的形式接收到查询参数q
的多个值:
{"q":["foo","bar"]}
要声明类型为list
的查询参数,需要显式地使用Query
,否则该参数将被解释为请求体
具有默认值的查询参数列表/多个值
可以给定默认list
值:
from typing import Listfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: List[str] = Query(default=["foo", "bar"])):query_items = {"q": q}return query_items
q: List[str] = Query(default=["foo", "bar"])
也可以使用list代替,这样不会检查列表的内容,如q: list = Query(default=[])
。
声明更多元数据
你可以添加更多有关该参数的信息。
这些信息将包含在生成的OpenAPI模式中,并由文档用户界面和外部工具所使用。
- 添加
title
:
from typing import Unionfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, title="Query string", min_length=3),
):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results
- 添加
description
:
from typing import Unionfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None,title="Query string",description="Query string for the items to search in the database that have a good match",min_length=3,),
):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results
别名参数(alias)
假设你想要查询参数为item-query
,如下:
http://127.0.0.1:8000/items/?item-query=foobaritems
但是item-query
不是一个有效的Python变量名称,这时可以用alias
参数声明一个别名,该别名将用于在URL中查找查询参数值:
from typing import Unionfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, alias="item-query")):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results
弃用参数(deprecated)
假设你不再喜欢该参数。
你不得不将其保留一段时间,因为有些客户端正在使用它,但你希望文档清楚地将其展示为已弃用。
那么将deprecated=True
传入Query
:
from typing import Unionfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None,alias="item-query",title="Query string",description="Query string for the items to search in the database that have a good match",min_length=3,max_length=50,pattern="^fixedquery$",deprecated=True,),
):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results
文档会像下面这样展示它:
总结
你可以为查询参数声明额外的校验和元数据。
通用的校验和元数据:
alias
title
description
deprecated
特定于字符串的校验:
min_length
max_length
regex
路径参数和数值校验
与使用Query
为查询参数声明更多的校验和元数据的方式相同,你也可以使用Path
为路径参数声明相同类型的校验和元数据。
具体步骤如下:
- 导入Path:首先从
fastapi
导入Path
- 声明元数据:可以声明与
Query
相同的所有参数,例如,要声明路径参数item_id
的title
元数据值,可以输入item_id: Annotated[int, Path(title="The ID of the item to get")]
from typing import Annotatedfrom fastapi import FastAPI, Path, Queryapp = FastAPI()@app.get("/items/{item_id}")
async def read_items(item_id: Annotated[int, Path(title="The ID of the item to get")],q: Annotated[str | None, Query(alias="item-query")] = None,
):results = {"item_id": item_id}if q:results.update({"q": q})return results
路径参数总是必需的。
按需对参数排序
假设你想要声明一个必需的str
类型查询参数q
。
而且你不需要为该参数声明任何其他内容,所以实际上并不需要使用Query
。
但是你仍然需要使用Path
来声明路径参数item_id
。
如果你将带有【默认值】的参数放在没有【默认值】的参数之前,Python将会报错。
但是你可以对其重新排序,并将不带默认值的值(查询参数q
)放到最前面。
对FastAPI来说这无关紧要。它将通过参数的名称、类型和默认值声明(Query
、Path
等)来检测参数,而不在乎参数的顺序。
因此可以将函数声明为:
from fastapi import FastAPI, Pathapp = FastAPI()@app.get("/items/{item_id}")
async def read_items(q: str, item_id: int = Path(title="The ID of the item to get")):results = {"item_id": item_id}if q:results.update({"q": q})return results
还可以用*
表示后面的所有参数作为关键字参数,也被称为kwargs
来调用,即使它们没有默认值:
from fastapi import FastAPI, Pathapp = FastAPI()@app.get("/items/{item_id}")
async def read_items(*, item_id: int = Path(title="The ID of the item to get"), q: str):results = {"item_id": item_id}if q:results.update({"q": q})return results
数值校验
使用Query
和Path
(以及后面的其他类)可以声明字符串约束,也可以声明数值约束。如下,添加ge=1
后,item_id
将必须是一个大于(greater than
)或等于(equal
)1
的整数。
- 大于等于
from fastapi import FastAPI, Pathapp = FastAPI()@app.get("/items/{item_id}")
async def read_items(*, item_id: int = Path(title="The ID of the item to get", ge=1), q: str
):results = {"item_id": item_id}if q:results.update({"q": q})return results
- 大于:
gt
(greater than) - 小于等于:
le
(less than or equal)
from fastapi import FastAPI, Pathapp = FastAPI()@app.get("/items/{item_id}")
async def read_items(*,item_id: int = Path(title="The ID of the item to get", gt=0, le=1000),q: str,
):results = {"item_id": item_id}if q:results.update({"q": q})return results
- 浮点数:数值校验同样适用于
float
值。
请求体-多个参数
既然我们已经知道了如何使用Path
和Query
,下面让我们来了解一下请求体声明的更高级用法。
混合使用Path
、Query
和请求体参数
你可以随意混合使用Path
、Query
和请求体参数声明。
你还可以通过将默认值设置为None
来将请求体参数声明为可选参数:
from typing import Annotatedfrom fastapi import FastAPI, Path
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = None@app.put("/items/{item_id}")
async def update_item(item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)],q: str | None = None,item: Item | None = None,
):results = {"item_id": item_id}if q:results.update({"q": q})if item:results.update({"item": item})return results
多个请求体参数
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneclass User(BaseModel):username: strfull_name: str | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):results = {"item_id": item_id, "item": item, "user": user}return results
在上面的情况下,FastAPI将注意到该函数中有多个请求体参数(两个Pydantic模型参数)。因此,它将使用参数名称作为请求体中的键(字段名称),并期望一个类似于以下内容的请求体:
{"item": {"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2},"user": {"username": "dave","full_name": "Dave Grohl"}
}
FastAPI将自动对请求中的数据进行转换,因此item
参数将接收指定的内容,user
参数也是如此。它将执行对复合数据的校验,并且像现在这样为OpenAPI模式和自动化文档对其进行记录。
请求体中的单一值(Body)
与使用Query
和Path
为查询参数和路径参数定义额外数据的方式相同,FastAPI提供了一个同等的Body
,例如上面的模型,除了item
和user
之外,还想在同一请求体中具有另一个键importance
,如果按原样来声明它,因为它是一个单一值,FastAPI将假定它是一个查询参数。但是你可以使用Body
指示FastAPI将其作为请求体的另一个键进行处理。
from typing import Annotatedfrom fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneclass User(BaseModel):username: strfull_name: str | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):results = {"item_id": item_id, "item": item, "user": user, "importance": importance}return results
在这种情况下,FastAPI将期望这样的请求体:
{"item": {"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2},"user": {"username": "dave","full_name": "Dave Grohl"},"importance": 5
}
多个请求体参数和查询参数
除了请求体参数外,你还可以在任何需要的时候声明额外的查询参数。
由于默认情况下单一值被解释为查询参数,因此你不必显式地添加Query
。
from typing import Annotatedfrom fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneclass User(BaseModel):username: strfull_name: str | None = None@app.put("/items/{item_id}")
async def update_item(*,item_id: int,item: Item,user: User,importance: Annotated[int, Body(gt=0)],q: str | None = None,
):results = {"item_id": item_id, "item": item, "user": user, "importance": importance}if q:results.update({"q": q})return results
嵌入单个请求体参数(Body(embed=True))
假设你只有一个来自Pydantic模型Item
的请求体参数item
。
默认情况下,FastAPI将直接期望这样的请求体。
但是,如果你希望它期望一个拥有item
键并在值中包含模型内容的JSON,就像在声明额外的请求体参数时所做的那样,则可以使用一个特殊的Body
参数embed
:
item: Item = Body(embed=True)
from typing import Annotatedfrom fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):results = {"item_id": item_id, "item": item}return results
在这种情况下,FastAPI将期望像这样的请求体:
{"item": {"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2}
}
而不是:
{"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2
}
总结
你可以添加多个请求体参数到路径操作函数中,即使一个请求只能有一个请求体。
但是 FastAPI 会处理它,在函数中为你提供正确的数据,并在路径操作中校验并记录正确的模式。
你还可以声明将作为请求体的一部分所接收的单一值。
你还可以指示 FastAPI 在仅声明了一个请求体参数的情况下,将原本的请求体嵌入到一个键中。
请求体-字段(Field)
与在路径操作函数中使用Query
、Path
、Body
声明校验与元数据的方式一样,可以使用Pydantic的Field
在Pydantic模型内部声明校验和元数据。
具体步骤如下:
- 导入
Field
:首先从Pydantic中导入Field
。 - 声明模型属性:使用
Field
定义模型的属性。
from typing import Annotatedfrom fastapi import Body, FastAPI
from pydantic import BaseModel, Fieldapp = FastAPI()class Item(BaseModel):name: strdescription: 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: float | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):results = {"item_id": item_id, "item": item}return results
Field
的工作方式和Query
、Path
、Body
相同,参数也相同。
请求体-嵌套类型
使用FastAPI,你可以定义、校验、记录文档并使用任意深度嵌套的模型(归功于Pydantic)。
List字段
你可以将一个属性定义为拥有子元素的类型。例如Pythonlist
:
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: list = []@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):results = {"item_id": item_id, "item": item}return results
这将使tags
成为一个由元素组成的列表。不过它没有声明每个元素的类型。
具有子类型的List字段
但是Python有一种特定的方法来声明具有子类型的列表:
具体步骤如下:
- 声明具有子类型的List:从
typing
模块导入它们,使用方括号[
和]
将子类型作为【类型参数】传入。 - 在我们的示例中,我们可以将
tags
明确地指定为一个【字符串列表】。
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: list[str] = []@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):results = {"item_id": item_id, "item": item}return results
Set类型
标签不应该重复,所以用Set
将tag
声明为一个由str
组成的set
:
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: set[str] = set()@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):results = {"item_id": item_id, "item": item}return results
这样,即使收到带有重复数据的请求,这些数据也会被转换为一组唯一项。
而且,每当你输出该数据时,即使源数据有重复,它们也将作为一组唯一项输出。
并且还会被相应地标注/记录文档。
嵌套类型
Pydantic模型的每个属性都具有类型。但是这个类型本身可以是另一个Pydantic模型。因此,你可以声明拥有特定属性名称、类型和校验的深度嵌套的JSON对象。
例如:
- 定义一个
Image
模型 - 将
Image
模型用作一个属性的类型
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Image(BaseModel):url: strname: strclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: set[str] = set()image: Image | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):results = {"item_id": item_id, "item": item}return results
这意味着FastAPI将期望类似于以下内容的请求体:
{"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2,"tags": ["rock", "metal", "bar"],"image": {"url": "http://example.com/baz.jpg","name": "The Foo live"}
}
特殊的类型和校验
除了普通的单一值类型(如str
、int
、float
等)外,你还可以使用从str
继承的更复杂的单一值类型。
要了解所有的可用选项,查看来自Pydantic的外部类型的文档。
例如,在Image
模型中我们有一个url
字段,我们可以把它声明为Pydantic的HttpUrl
,而不是str
:
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrlapp = FastAPI()class Image(BaseModel):url: HttpUrlname: strclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: set[str] = set()image: Image | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):results = {"item_id": item_id, "item": item}return results
该字符串将被检查是否为有效的URL,并在JSON Schema/OpenAPI文档中进行记录。
带有一组子模型的属性
你还可以将Pydantic模型用作list
、set
等的子模型:
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrlapp = FastAPI()class Image(BaseModel):url: HttpUrlname: strclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: set[str] = set()images: list[Image] | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):results = {"item_id": item_id, "item": item}return results
这将期望(转换,校验,记录文档等)下面这样的JSON请求体:
{"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2,"tags": ["rock","metal","bar"],"images": [{"url": "http://example.com/baz.jpg","name": "The Foo live"},{"url": "http://example.com/dave.jpg","name": "The Baz"}]
}
深度嵌套模型
你可以定义任意深度的嵌套模型:
from typing import Unionfrom fastapi import FastAPI
from pydantic import BaseModel, HttpUrlapp = FastAPI()class Image(BaseModel):url: HttpUrlname: strclass Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: set[str] = set()images: Union[list[Image], None] = Noneclass Offer(BaseModel):name: strdescription: Union[str, None] = Noneprice: floatitems: list[Item]@app.post("/offers/")
async def create_offer(offer: Offer):return offer
纯列表请求体
如果你期望的JSON请求体的最外层是一个JSONarray
(即Python list
),则可以在路径操作函数的参数中声明此类型,就像声明Pydantic模型一样:
images: List[Image]
例如:
from 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
任意dict
构成的请求体
你也可以将请求体声明为使用某类型的键和其他类型值的dict
。
无需事先知道有效的字段/属性(在使用Pydantic模型的场景)名称是什么。
如果你想接收一些尚且未知的键,这将很有用。
其他有用的场景是当你想要接收其他类型的键时,例如int
,如下,你将接受任意键为int
类型并且值为float
类型的dict
:
from fastapi import FastAPIapp = FastAPI()@app.post("/index-weights/")
async def create_index_weights(weights: dict[int, float]):return weights
JSON仅支持将str
作为键,但是Pydantic具有自动转换数据的功能。这意味着,即使你的API客户端只能将字符串作为键发送,只要这些字符串内容仅包含整数,Pydantic就会对其进行转换并校验。然后你接收的名为weights
的dict
实际上将具有int
类型的键和float
类型的值。
模式的额外信息-例子
你可以在JSON模式中定义额外的信息。
一个常见的用例是添加一个将在文档中显示的example
。
有几种方法可以声明额外的JSON模式信息。
Pydantic schema_extra
你可以使用Config
和schema_extra
为Pydantic模型声明一个示例,如Pydantic文档:定制Schema中所述。
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonemodel_config = {"json_schema_extra": {"examples": [{"name": "Foo","description": "A very nice Item","price": 35.4,"tax": 3.2,}]}}@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):results = {"item_id": item_id, "item": item}return results
这些额外的信息将按原样添加到输出的JSON模式中。
Field
的附加参数
在Field
,Path
,Query
,Body
和其他你之后将会看到的工厂函数,你可以为JSON模式声明额外信息,你也可以通过给工厂函数传递其他的任意参数来给JSON模式声明额外信息,比如增加example
:
from fastapi import FastAPI
from pydantic import BaseModel, Fieldapp = FastAPI()class Item(BaseModel):name: str = Field(examples=["Foo"])description: str | None = Field(default=None, examples=["A very nice Item"])price: float = Field(examples=[35.4])tax: float | None = Field(default=None, examples=[3.2])@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):results = {"item_id": item_id, "item": item}return results
传递的那些额外参数不会添加任何验证,只会添加注释,用于文档的目的。
Body
额外参数
from typing import Annotatedfrom fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int,item: Annotated[Item,Body(examples=[{"name": "Foo","description": "A very nice Item","price": 35.4,"tax": 3.2,}],),],
):results = {"item_id": item_id, "item": item}return results
额外数据类型
到目前为止,一直在使用常见的数据类型,如:
int
float
str
bool
但是也可以使用更复杂的数据类型。
其他数据类型
UUID
:
一种标准的 “通用唯一标识符” ,在许多数据库和系统中用作ID。
在请求和响应中将以str
表示。datetime.datetime
:
一个 Pythondatetime.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
:
标准的 Pythonbytes
。
在请求和响应中被当作str
处理。
生成的模式将指定这个str
是binary
“格式”。Decimal
:
标准的 PythonDecimal
。
在请求和响应中被当做float
一样处理。
您可以在这里检查所有有效的pydantic数据类型: Pydantic data types.
例子
from datetime import datetime, time, timedelta
from typing import Annotated
from uuid import UUIDfrom fastapi import Body, FastAPIapp = FastAPI()@app.put("/items/{item_id}")
async def read_items(item_id: UUID,start_datetime: Annotated[datetime, Body()],end_datetime: Annotated[datetime, Body()],process_after: Annotated[timedelta, Body()],repeat_at: Annotated[time | None, Body()] = None,
):start_process = start_datetime + process_afterduration = end_datetime - start_processreturn {"item_id": item_id,"start_datetime": start_datetime,"end_datetime": end_datetime,"process_after": process_after,"repeat_at": repeat_at,"start_process": start_process,"duration": duration,}