avatar


5.FastAPI [1/3]

什么是FastAPI

FastAPI,轻量级的Web框架。

安装

需要安装FastAPI以及Uvicorn

FastAPI

FastAPI的安装命令:

1
pip instal1 fastapi

pip工具在安装Pyhton库的时候,会自动检查并安装依赖库。这里核心的依赖库包括PydanticStarlette等。

Uvicorn

Uvicorn是一个ASGI(Asynchronous Server Gateway Interface,异步服务器网关接口)服务器框架,Uvicorn为FastAPI提供了快速异步运行环境功能。

安装命令:

1
pip install uvicorn

入门案例

代码

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 导入 FastAPI类
from fastapi import FastAPI
# 导入uvicorn,ASGI服务器
import uvicorn

# 创建应用实例
app = FastAPI()


@app.get("/")
# 定义路由路径
async def root():
# 定义路径操作函数
return {"msg": "hello"}


if __name__ == '__main__':
uvicorn.run(app=app)

运行结果:

1
2
3
4
5
INFO:     Started server process [6616]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: 127.0.0.1:50538 - "GET / HTTP/1.1" 200 OK

然后我们访问http://127.0.0.1:8000,会收到返回如下:

1
{"msg":"hello"}

特别的,FastAPI已经集成了Swagger。通过http://127.0.0.1:8000/docs,即可访问Swagger文档;通过http://127.0.0.1:8000/redoc,可以访问另一种风格的Swagger文档。

如果我们把函数定义为def而不是async def,那么FastAPI会把它放到单独的线程池中,异步执行。也就是说无论是否使用async,FastAPI都将异步工作,以达到"Fast"的运行速度。

问题解决

我们可能会遇到如下的报错:

1
ImportError: cannot import name Deque

这个可能因为我们的Python版本小于3.6.1,需要3.6.1及以上版本的Python。

框架构成

框架功能

框架功能

当用户通过浏览器发起请求数据时,FastAPI服务器端对请求数据做以下操作:

  1. 由FastAPI中间件接收请求数据,对数据进行初步的处理。
  2. 将请求的URL中的路径与FastAPI定义的路由列表进行匹配。
  3. FastAPI对请求数据进行数据验证和数据转换,得到符合要求的数据,并将数据传递给路径操作函数。
  4. 路径操作函数接收请求数据后,调用"业务处理"对数据进行加工、对资源进行读写,再将处理结果封装成响应数据。
  5. 将响应数据传递给FastAPI中间件,由FastAPI中间件对数据进行再次处理后,返回给浏览器。

在一些关键技术方面:

  • FastAPI以Starlette库作为Web服务器的底层,提供了异步技术接收客户端发起的请求数据。
  • 通过高性能的数据模型框架Pydantic库对数据进行验证和转换,响应数据也通过Pydantic库转换成符合JSON模式的响应数据。

Pydantic

Pydantic,基于Python的类型注解的数据模型定义及验证框架。

基本用法

Pydantic中使用自定义类的方式定义数据模型类,而且数据模型类必须继承Pydantic的BaseModel。

数据模型的基本用法,示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from pydantic import BaseModel


# 从 pydantic模块中导人BaseModel类

# 继承BaseModel类,定义数据模型 User类
class User(BaseModel):
# 字段id类型为int
id: int
# 字段name未指定类型,值为Kaka
name = 'Kaka'


u = User(id="123")

# 实例 u 中字段id的值
print(u.id)
# 实例 u 中字段name的值
print(u.name)

# 实例 u 包含的字段集合
print(u.__fields_set__)
# 实例 u 的dict()方法
print(u.dict())

uu = User(id='aaa')

运行结果:

1
2
3
4
123
Kaka
{'id'}
{'id': 123, 'name': 'Kaka'}

在上述代码中,从Pydantic模块导入了BaseModel类,定义了一个数据模型类User并继承了BaseModel类,数据模型类中有两个属性id和name。字段id类型为int,是必填项,字段name未指定数据类型,并且初始值为"Kaka"。

特别的,如果给id赋值为非数字,会报错,示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from pydantic import BaseModel


# 从 pydantic模块中导人BaseModel类

# 继承BaseModel类,定义数据模型 User类
class User(BaseModel):
# 字段id类型为int
id: int
# 字段name未指定类型,值为Kaka
name = 'Kaka'


u2 = User(id='aaa')
print(u2.id)

运行结果:

1
2
3
4
5
6
7
Traceback (most recent call last):
File "C:\Dev\f\main.py", line 14, in <module>
u2 = User(id='aaa')
File "pydantic\main.py", line 341, in pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for User
id
value is not a valid integer (type=type_error.integer)

解释说明:Pydantic在验证数据时,对无效的数据会进行错误提示。

属性和方法

Pydantic模型类为定义的数据模型类实例提供如下方法和属性:

  1. dict(),将数据模型的字段和值封装成字典。
  2. json(),将数据模型的字段和值封装成JSON格式字符串。
  3. copy(),生成数据模型实例的副本。
  4. parse_obj(),将Python字典数据解析为数据模型实例。
  5. parse_raw(),将字符串解析为数据模型实例。
  6. parse_file(),传入文件路径,并将路径所对应的文件解析为数据模型实例。
  7. from_orm(),将任何自定义类的实例转换成数据模型对象。
  8. schema(),将数据模型转换成JSON模式数据。
  9. schema_json(),返回schema()生成的字符串。
  10. construct()类方法,创建数据模型实例时不进行验证。
  11. __fields_set__属性,创建数据模型实例的初始化字段列表。
  12. __fields__属性,罗列数据模型的全部字段的字典。
  13. __config__属性,显示数据模型的配置类。

嵌套模型

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from typing import List
from pydantic import BaseModel


# 定义数据模型类继承自BaseModel
class Blackboard(BaseModel):
# 字段类型 int
size = 4000
# 字段类型 str
color: str


# 定义数据模型类继承自 BaseModel
class Table(BaseModel):
# 字段默认为空白
position: str


# 定义数据模型类继承自 BaseModel
class ClassRoom(BaseModel):
# 字段类型使用数据模型类Body
blackboard: Blackboard
# 字段类型使用列表泛型
tables: List[Table]


# 创建数据模型实例
m = ClassRoom(
blackboard={'color': 'green'},
tables=[{'position': '第一排左1'}, {'position': '第一排左2'}]
)

print(m)
print(m.dict())

运行结果:

1
2
blackboard=Blackboard(color='green', size=4000) tables=[Table(position='第一排左1'), Table(position='第一排左2')]
{'blackboard': {'color': 'green', 'size': 4000}, 'tables': [{'position': '第一排左1'}, {'position': '第一排左2'}]}

在上述代码中,首先定义了两个数据模型类BlackBoard和Table,第三个数据模型ClassRoom的字段类型分别定义为前两个数据模型类。之后创建了数据模型ClassRoom的实例m,并打印出实例m的数据,数据中包含了数据模型全部的字段和值。

Starlette

Starlette是一个轻量级的、高性能异步服务网关接口框架(ASGI)。FastAPI框架中高性能异步操作的特性主要来源于Starlette。

Starlette的使用,示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
import uvicorn


# 定义异步函数,参数为请求对象
async def root(request):
# 返回 JSON 响应
return JSONResponse({'msg': 'hello'})


# 创建服务实例
# 定义路由,将路径/指向异步函数
app = Starlette(debug=True, routes=[Route('/', root)])

if __name__ == '__main__':
# 使用 uvicorn 启动服务器
uvicorn.run(app=app)

运行结果:

1
2
3
4
INFO:     Started server process [2256]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

然后我们访问http://127.0.0.1:8000,会收到返回如下:

1
{"msg":"hello"}

在上述代码中,使用async def关键字定义了一个异步函数root,函数返回了一个响应对象(JSONResponse);然后创建了一个Starlette实例,实例中传入了路由列表,列表中第一个路由对象将路径/指向了root路径操作函数,路径操作函数将响应对象返回给浏览器。

在上文的FastAPI的例子中,我们没有定义路由列表,而是使用装饰器的方式设置"路由路径"与路径操作函数的绑定关系。但,其实,FastAPI在运行时,还是会在内部建立"路由表",将装饰器设置的"路由"保存到"路由表"中,然后启动服务。FastAPI只是给我们提供了一种更简洁的方式。

请求

路径参数

所谓的路径参数,即REST风格、RESTful。

关于REST风格、RESTful,可以参考《基于Java的后端开发入门:17.SpringMVC》的"REST风格"部分,本文不赘述。

简单路径参数

在FastAPI中需要先用带装饰符{}@app.get("/items/{id_value}")方法注册路由,用方法中的参数id_value接收URL地址传递过来的路径参数值,同时,路径操作函数的参数名需要与上一行路由中的路径参数名一致。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from fastapi import FastAPI
import uvicorn

# 创建应用实例
app = FastAPI()


# 注册路由路径,使用{}定义路径参数,参数名为item_id
@app.get("/items/{item_id}")
# 在路径操作函数中定义同名的路径参数
async def read_item(item_id):
print(item_id)
return {"item_id": item_id}


if __name__ == '__main__':
uvicorn.run(app=app)

访问http://127.0.0.1:8000/items/123,返回如下:

1
{"item_id":"123"}

有类型的路径参数

有类型的路径参数,指的是在定义路径操作函数的参数时,需要指定数据类型,如整型、字符串等。

具体通过类型注解指定参数的数据类型。

如果我们在参数定义时,没有指定数据类型,则会默认为str类型。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from fastapi import FastAPI
import uvicorn

# 创建应用实例
app = FastAPI()


# 注册路由路径,使用{}定义路径参数,参数名为item_id
@app.get("/items/{item_id}")
# 在路径操作函数中定义同名的路径参数
# 【整型 int】
async def read_item(item_id: int):
print(item_id)
return {"item_id": item_id}


if __name__ == '__main__':
uvicorn.run(app=app)

访问http://127.0.0.1:8000/items/123,返回如下:

1
{"item_id":123}

特别的,如果我们传入非数字的路径参数,例如abc,访问http://127.0.0.1:8000/items/abc,会收到报错,返回如下:

1
{"detail":[{"loc":["path","item_id"],"msg":"value is not a valid integer","type":"type_error.integer"}]}

使用枚举类型参数

我们可以使用枚举(Enum),对路径参数中接收到的值进行验证,并转换为枚举类型的数据。
在FastAPI中使用枚举类型定义参数的方法,示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from enum import Enum
from fastapi import FastAPI
import uvicorn


# 定义枚举类,继承 str,Enum
class ModelName(str, Enum):
a = "aaa"
b = "bbb"
c = "ccc"


app = FastAPI()


# 注册路由路径,包含路径参数 model_name
@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
print(model_name)
return {"model_name": model_name}


if __name__ == '__main__':
uvicorn.run(app=app)

访问http://127.0.0.1:8000/models/aaa,返回如下:

1
{"model_name":"aaa"}

解释说明:定义了枚举类ModelName,继承str类型和枚举类型(Enum)。

特别的,我们可以看看API文档中的Available values

使用枚举类型参数

路由访问顺序

在FastAPI中使用装饰器注册路由路径时,路由路径按照代码中的顺序保存到后端服务器的路由表中,如果后端服务器注册了多个路由路径时,则涉及到匹配顺序要求。

基本的匹配访问原则为从上到下。

题外话
在Express中,匹配顺序也是按照路由注册的顺序进行匹配。
关于Express,可以参考《基于JavaScript的前端开发入门:3.Node.js》的"express"部分。

在下文的例子中,注册了2个路由,第一个是静态的路由路径"/users/me",第二个是定义了带路径参数的路由路径"/users/{user_id}“。
访问URL/users/me,按照顺序,会与第一个”/users/me"进行匹配,不与之后的进行匹配。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from fastapi import FastAPI
import uvicorn

app = FastAPI()


@app.get("/users/me")
async def read_users_me():
print('read_users_me')
return {"user_id": 'read_users_me'}


@app.get("/users/{user_id}")
async def read_user(user_id: str):
print('read_user')
return {"user_id": user_id}


if __name__ == '__main__':
uvicorn.run(app=app)

访问http://127.0.0.1:8000/users/me,返回如下:

1
{"user_id":"read_users_me"}

同时后台打印如下:

1
read_users_me

访问http://127.0.0.1:8000/users/some,返回如下:

1
{"user_id":"some"}

同时后台打印如下:

1
read_user

特别的,如果我们调整@app.get("/users/me")@app.get("/users/{user_id}")的顺序,访问http://127.0.0.1:8000/users/me,返回如下:

1
{"user_id":"me"}

同时后台打印如下:

1
read_user

查询参数

标准查询参数

本文的"标准查询参数",是指类似如下的查询方式,其中p为键、v为值。当出现两个及以上的查询参数时,参数之间用&关联。

1
?p1=v1&p2=v2&...

例如,http://127.0.0.1:8000/items?skip=0&limit=10,则?后跟着skip=0limit=10两个查询参数。

在路径操作函数里,通过定义函数参数的方式定义查询参数,与路由路径注册无关。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from fastapi import FastAPI
import uvicorn

app = FastAPI()

# 定义列表items
items = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']


# 注册路由路径,未定义路径参数
@app.get("/items/")
# 定义了两个参数,参数类型为 int
async def read_item(skip: int = 0, limit: int = 10):
print('参数 skip:', skip)
print('参数 limit:', limit)
# 用下标方式从列表items 中取出数据
return items[skip:skip + limit]


if __name__ == '__main__':
uvicorn.run(app=app)

访问http://127.0.0.1:8000/items/?skip=1&limit=2,返回如下:

1
["b","c"]

后台打印日志如下:

1
2
参数 skip: 1
参数 limit: 2

在上例中,路径操作函数read_item定义了两个参数skiplimit,这样定义的参数称为查询参数。

查询参数其实是URL地址的一部分,因此"原始输入值"是字符串;在本例中,因为在路径操作函数里定义时,显式指定了int,FastAPI自动将其转换为整型。

可选查询参数

可选查询参数,在路径操作函数里用Optional关键字定义。

在下文的例子中,在路径操作函数read_item中使用Optional关键字定义了一个可选参数q,如果没有传递该参数,其值为None。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from typing import Optional
import uvicorn
from fastapi import FastAPI

app = FastAPI()


# 注册路由路径,定义路径参数
@app.get('/items/{item_id}')
# 定义路径操作函数
# 定义路径参数item_id,参数类型为 str
# 定义可选查询参数q,参数类型为 str,默认值为 None
async def read_item(item_id: str, q: Optional[str] = None):
if q:
return {'item_id': item_id, 'q': q}
else:
return {'item_id': item_id}


if __name__ == '__main__':
uvicorn.run(app=app)

访问http://127.0.0.1:8000/items/123?q=abc,返回如下:

1
{"item_id":"123","q":"abc"}

访问http://127.0.0.1:8000/items/12345,返回如下:

1
{"item_id":"12345"}

必选查询参数

在路径操作函数中定义带常规类型的查询参数,并且不带默认值,则该参数是必选参数。

在下文代码中,函数read_item定义了两个必选查询参数,这两个参数都没有指定默认值,其中item_id在装饰器中定义为路径参数,是必传的;q未在装饰器中定义,所以是查询参数。这两个参数的值都是必须传的,如果不传将会导致验证错误。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import uvicorn
from fastapi import FastAPI

app = FastAPI()


# 注册路由路径,定义路径参数
@app.get('/items/{item_id}')
# 定义路径操作函数
# 定义路径参数item_id,参数类型为 str
# 定义查询参数q,参数类型为 str
async def read_item(item_id: str, q: str):
if q:
return {'item_id': item_id, 'q': q}
else:
return {'item_id': item_id}


if __name__ == '__main__':
uvicorn.run(app=app)

访问http://127.0.0.1:8000/items/123?q=abc,返回如下:

1
{"item_id":"123","q":"abc"}

访问http://127.0.0.1:8000/items/123,返回如下,缺少必选参数。

1
{"detail":[{"loc":["query","q"],"msg":"field required","type":"value_error.missing"}]}

参数类型转换(布尔类型)

关于"参数类型转换",其实在上文我们已经提到了。通过URL传递的查询参数,参数值的原始类型是字符串,在上文中,将参数类型定义为int类型,FastAPI会验证参数并将其转换为int类型。

在这里我们讨论FastAPI解析bool类型。

  • Truetrueyes1,会转换成布尔值True。
  • Falsefalseno0,会转换成布尔值False。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from typing import Optional
import uvicorn
from fastapi import FastAPI

app = FastAPI()


# 注册路由路径,定义路径参数item_id
@app.get("/items/{item_id}")
# 定义路径操作函数
# 定义路径参数,类型为 str
# 定义可选查询参数q,类型 str
# 定义查询参数,类型为 bool,默认值为 False
async def read_item(item_id: str, q: Optional[str] = None, short: bool = False):
# 创建对象item,赋值路径参数item_id
item = {"item_id": item_id}

if q:
# 当传入了可选查询参数q时,更新item
item.update({"q": q})
if short:
# 当传入了查询参数short,并且其值为True时,更新数据
item.update({"description": "描述描述"})

return item


if __name__ == '__main__':
uvicorn.run(app=app)

访问http://127.0.0.1:8000/items/123,返回如下:

1
{"item_id":"123"}

访问http://127.0.0.1:8000/items/123?short=True,返回如下:

1
{"item_id":"123","description":"描述描述"}

访问http://127.0.0.1:8000/items/123?short=true,返回如下:

1
{"item_id":"123","description":"描述描述"}

访问http://127.0.0.1:8000/items/123?short=yes,返回如下:

1
{"item_id":"123","description":"描述描述"}

访问http://127.0.0.1:8000/items/123?short=1,返回如下:

1
{"item_id":"123","description":"描述描述"}

请求体

在本文,我们讨论如何在FastAPI中接收请求体。

定义请求体的数据模型

我们通过继承Pydantic的BaseModel,定义请求体的数据模型。

在下文的例子中,定义了一个继承BaseModel的数据模型类Item。
Item类中定义了4个字段,其中name和price是必选参数,另外两个参数description和tax使用了Optional,是可选参数。

在注册路由路径时,需要使用@app.post装饰器,即将请求方法设置为POST,以支持请求体的传递。
@app.get装饰器注册路由,不支持传递请求体。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from typing import Optional
import uvicorn
from fastapi import FastAPI
# 导入基础模型类
from pydantic import BaseModel


# 定义数据模型类,继承自BaseModel类
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None


app = FastAPI()


# 注册路由路径,请求方式为post,不是get
@app.post("/items/")
# 定义请求体,类型是数据模型类Item
# 直接返回请求体数据
async def create_item(item: Item):
return item


if __name__ == '__main__':
uvicorn.run(app=app)

传送请求体,示例代码:

1
2
3
4
5
6
7
8
9
10
curl -X 'POST' \
'http://127.0.0.1:8000/items/' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"name": "名名名",
"description": "描述描述",
"price": 100,
"tax": 0.123
}'

运行结果:

1
2
3
4
5
6
{
"name": "名名名",
"description": "描述描述",
"price": 100,
"tax": 0.123
}

可选的请求体参数

上文,我们在查询参数中介绍了使用Optional关键字将参数设置为可选参数的方法。
类似的,我们也可以使用Optional关键字将请求体参数设置为可选参数。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from typing import Optional
import uvicorn
from fastapi import FastAPI, Path
from pydantic import BaseModel

app = FastAPI()


# 定义数据模型类,继承自BaseModel
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None


# 注册路由路径,使用PUT方法,定义路径参数
@app.put("/items/{item_id}")
# * python语法,表示后面的参数都是键值对
async def update_item(*,
item_id: int = Path(..., title="元素ID", ge=0, le=1000),
q: Optional[str] = None,
item: Optional[Item] = None):
results = {"item_id": item_id}
if q:
results.update({"q": q})
if item:
results.update({"item": item})
return results


if __name__ == '__main__':
uvicorn.run(app=app)

同时使用多个请求体

在下文的例子中,我们定义了两个数据模型,都继承了BaseModel。
在路径操作函数的参数列表中,也定义了两个请求体对象,分别对应两个数据模型。
Web后端服务器期望的请求体对象的格式,使用参数名(对应函数中的itemuser)作力JSON模式中的键,参数名对应的数据是每个键对应的JSON模式文本。
例如:

1
2
3
4
5
6
7
8
9
10
11
12
{
"item": {
"name": "名名",
"description": "描述",
"price": 123.45,
"tax": 0.123
},
"user": {
"username": "用户名",
"full_name": "完整名"
}
}

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from typing import Optional
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


# 定义数据模型Item,继承自BaseModel
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None


# 定义数据模型User,继承自BaseModel
class User(BaseModel):
username: str
full_name: Optional[str] = None


# 注册路由路径,使用PUT方法,定义路径参数item_id
@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


if __name__ == '__main__':
uvicorn.run(app=app)

同时传送多个请求体,示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
curl -X 'PUT' \
'http://127.0.0.1:8000/items/123' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"item": {
"name": "名名",
"description": "描述",
"price": 123.45,
"tax": 0.123
},
"user": {
"username": "用户名",
"full_name": "完整名"
}
}'

运行结果:

1
2
3
4
5
6
7
8
9
10
11
12
13
{
"item_id": 123,
"item": {
"name": "名名",
"description": "描述",
"price": 123.45,
"tax": 0.123
},
"user": {
"username": "用户名",
"full_name": "完整名"
}
}

常规数据类型作为请求体使用

在上文,定义请求体对象的方法是,在路径操作函数中将参数的数据类型设置成继承自Pydantic的BaseModel类。

在有些情况下,请求数据可能不是对象,而是一个Python常规数据类型的值,此时可以使用Body类管理这样的数据。
使用Body类,可以将指定参数设置为请求体对象中的另外一个键。

在下文的例子中,增加一个请求体参数importance,其数据类型为int。同时,在调用Body类时,使用了一个参数gt=0,这是Pydantic模型中的用法,可以给参数增加校验规则。gt=0的含义是importance的值要大于0。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from typing import Optional
import uvicorn
# 导入Body
from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


# 定义数据模型Item,继承自BaseModel
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None


# 定义数据模型User,继承自BaseModel
class User(BaseModel):
username: str
full_name: Optional[str] = None


# 注册路由路径,使用PUT方法,定义路径参数item_id
@app.put("/items/{item_id}")
# 定义路径操作函数
async def update_item(item_id: int, item: Item, user: User, importance: int = Body(..., gt=0)):
results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
return results


if __name__ == '__main__':
uvicorn.run(app=app)

发送请求提,示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
curl -X 'PUT' \
'http://127.0.0.1:8000/items/123' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"item": {
"name": "名名名名",
"description": "描述描述",
"price": 0,
"tax": 0
},
"user": {
"username": "用户名",
"full_name": "完整名"
},
"importance": 123
}'

运行结果:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
{
"item_id": 123,
"item": {
"name": "名名名名",
"description": "描述描述",
"price": 0,
"tax": 0
},
"user": {
"username": "用户名",
"full_name": "完整名"
},
"importance": 123
}

特别的,如果我们的importance小于等于0,会收到报错信息,示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
curl -X 'PUT' \
'http://127.0.0.1:8000/items/123' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"item": {
"name": "名名名名",
"description": "描述描述",
"price": 0,
"tax": 0
},
"user": {
"username": "用户名",
"full_name": "完整名"
},
"importance": 0
}'

运行结果:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
{
"detail": [
{
"loc": [
"body",
"importance"
],
"msg": "ensure this value is greater than 0",
"type": "value_error.number.not_gt",
"ctx": {
"limit_value": 0
}
}
]
}

同时使用路径参数、查询参数和请求体

我们可以同时使用路径参数、查询参数和请求体。
FastAPI会依次按照如下顺序对路径操作函数的参数进行解析:

  1. 在注册的路由路径中匹配参数名称,匹配到的参数会被解析为路径参数,未匹配到路径参数名称的进入第二步匹配。
  2. 如果参数属于Python的常规类型(strintfloatbool等),则参数被解析为查询参数。
  3. 如果参数类型是数据模型类,则参数被解析为请求体参数。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from typing import Optional
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel


# 定义数据模型类,继承自BaseModel类
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None


app = FastAPI()


# 注册路由,定义路径参数item_id
@app.post("/items/{item_id}")
# 定义路径操作函数
# 定义路径参数,类型int
# 定义请求体,类型是数据模型
# 定义可选查询参数,类型str
async def create_item(item_id: int, item: Item, q: Optional[str] = None):
# 将路径参数和请求体参数组合为数据对象
result = {"item_id": item_id, **item.dict()}
if q:
# 如果传入了查询参数,则更新查询参数
result.update({"q": q})
return result


if __name__ == '__main__':
uvicorn.run(app=app)

发送请求,示例代码:

1
2
3
4
5
6
7
8
9
10
curl -X 'POST' \
'http://127.0.0.1:8000/items/123?q=abc' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"name": "名名名名",
"description": "描述描述",
"price": 123.45,
"tax": 0.123
}'

运行结果:

1
2
3
4
5
6
7
8
{
"item_id": 123,
"name": "名名名名",
"description": "描述描述",
"price": 123.45,
"tax": 0.123,
"q": "abc"
}

表单和文件

python-multipart

如果要在FastAPI中操作表单数据,首先需要安装一个第三方库python-multipart,安装命令如下:

1
pip install python-multipart

表单数据

在下文的例子中,导入模块Form,定义路径操作函数的参数时,请求体参数的数据类型是str,初始值调用了Form(...)函数获取到的Form对象。

发送表单字段格式时,在HTTP的请求头中指定数据编码为application/x-www-form-urlencoded

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 导入Form对象
from fastapi import FastAPI, Form
import uvicorn

app = FastAPI()


# 注册路由路径
@app.post("/login/")
# 定义路径操作函数
async def login(username: str = Form(...), password: str = Form(...)):
# 处理登录的代码
if password == '123456':
return {"username": username}


if __name__ == '__main__':
uvicorn.run(app=app)

发送表单请求,示例代码:

1
2
3
4
5
curl -X 'POST' \
'http://127.0.0.1:8000/login/' \
-H 'accept: application/json' \
-H 'Content-Type: application/x-www-form-urlencoded' \
-d 'username=%E7%94%A8%E6%88%B7&password=%E5%AF%86%E7%A0%81'

运行结果:

1
2
3
{
"username": "用户"
}

文件上传

文件上传同样以来python-multipart

与使用Body类、Form类的方式类似,上传文件时,需要引入File类。

在下文的例子中,首先导入了File类对象,然后定义了两个路径操作函数,两个路径操作函数中的参数都调用File类获取File对象实例。
第一个路径操作函数中的参数类型是bytes,可以用来接收文件流,因为在HTTP中的文件上传使用的是文件流,FastAPI接收文件流时,使用的就是bytes数据类型。当上传的文件比较小的时候,可以使用这种方式接收上传文件。
第二个路径操作函数的参数类型是UploadFile,这个类型使用了一种名为"假脱机文件"的技术,当内存中的数据尺寸超过最大限制后,会将部分数据存储在磁盘中。也就是说,这种方式适合处理大文件。

需要注意的是,上传文件时使用的编码格式为multipart/form-data

在上传文件的同时,可以提交表单数据,但是不能提交请求体,因为请求体使用的格式为application/json

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from fastapi import FastAPI, File, UploadFile
import uvicorn

app = FastAPI()


# 注册路由路径
@app.post("/files/")
# 定义路径操作函数,第一种上传方式
async def create_file(file: bytes = File(...)):
return {"file_size": len(file)}


# 注册路由路径
@app.post("/uploadfile/")
# 定义路径操作函数,第二种上传方式
async def create_upload_file(file: UploadFile = File(...)):
return {"filename": file.filename}


if __name__ == '__main__':
uvicorn.run(app=app)

发送请求"/files/",示例代码:

1
2
3
4
5
curl -X 'POST' \
'http://127.0.0.1:8000/files/' \
-H 'accept: application/json' \
-H 'Content-Type: multipart/form-data' \
-F 'file=@2024-01-30.png;type=image/png'

运行结果:

1
2
3
{
"file_size": 70522
}

发送请求"/uploadfile/",示例代码:

1
2
3
4
5
curl -X 'POST' \
'http://127.0.0.1:8000/uploadfile/' \
-H 'accept: application/json' \
-H 'Content-Type: multipart/form-data' \
-F 'file=@2024-01-30.png;type=image/png'

运行结果:

1
2
3
{
"filename": "2024-01-30.png"
}

UploadFile

UploadFile还提供了其他属性和方法,用来获取原始上传文件的元数据。

主要属性有:

  1. filename,原始文件名。
  2. content_type,文件类型,比如image/jpeg
  3. file,文件对象。

主要方法有:

  1. write(data),写入str或bytes类型的数据。
  2. read(size),从文件对象中的当前位置开始,读取size大小的数据。
  3. seck(offset),将当前位置指向文件中指定的位置,一般配合read(size)方法使用。
  4. close(),关闭文件对象。

表单和多文件上传

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from fastapi import FastAPI, File, Form, UploadFile
import uvicorn

app = FastAPI()


# 注册路由路径
@app.post("/files/")
# 定义路径操作函数
async def create_file(file: UploadFile = File(...), file2: UploadFile = File(...), token: str = Form(...)):
return {
"token": token,
"fileb_content_type": file.filename,
"fileb_content_type": file.content_type,
}


if __name__ == '__main__':
uvicorn.run(app=app)

发送请求,示例代码:

1
2
3
4
5
6
7
curl -X 'POST' \
'http://127.0.0.1:8000/files/' \
-H 'accept: application/json' \
-H 'Content-Type: multipart/form-data' \
-F 'file=@2024-01-30.png;type=image/png' \
-F 'file2=@2024-01-30.png;type=image/png' \
-F 'token=123456'

运行结果:

1
2
3
4
{
"token": "123456",
"fileb_content_type": "image/png"
}

响应

响应模型

响应模型(Response Model),指在处理响应数据时,将响应数据转换成Pydantic数据模型实例,以保证响应数据的规范性。同时,响应数据模型在API文档中体现为JSON模式,这样也增加了文档的可读性及接口的标准化。

类似我们在《基于Java的后端开发入门:18.SSM》所讨论的"结果封装"。

自定义响应模型

自定义响应模型,步骤如下:

  1. 定义响应数据模型(Response Data Model)类。
  2. 注册路由路径的装饰器方法中通过response_model参数指定响应数据模型来确定响应模型对象。
  3. 响应数据被转换为响应数据模型格式返回给客户端

在下文的例子中,定义了两个数据模型,请求数据模型UserIn、响应数据模型UserOut。其中UserOut少了password字段。
在路径操作函数的装饰器中,使用参数response_model定义响应模型为UserOut,所以FastAPI会使用UserOut模型将返回数据转换为响应数据,响应数据中将不包含password字段。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from typing import Optional
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


# 定义数据模型
class UserIn(BaseModel):
username: str
password: str
email: str
full_name: Optional[str] = None


# 定义数据模型
class UserOut(BaseModel):
username: str
email: str
full_name: Optional[str] = None


# 注册路由路径,定义响应模型为UserOut
@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
return user


if __name__ == '__main__':
uvicorn.run(app=app)

发送请求,示例代码:

1
2
3
4
5
6
7
8
9
10
curl -X 'POST' \
'http://127.0.0.1:8000/user/' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"username": "用户名",
"password": "密码",
"email": "邮箱",
"full_name": "全名"
}'

运行结果:

1
2
3
4
5
{
"username": "用户名",
"email": "邮箱",
"full_name": "全名"
}

自定义控制规则

在使用Pydantic定义数据模型时,允许字段有默认值。如果在输出响应数据时,这些带有默认值的字段没有明确设置字段值,客户端也可以按照默认值接收这些字段的数据。
若不想让客户端接收数据模型的默认值字段,则在路径操作函数的装饰器中设置响应模型参数时,可以同时设置response_model_exclude_unset参数,以忽略带有默认值的字段。

在下文的例子中,在路径操作函数read_item的装饰器参数里,设置了响应模型为Data,该数据模型中包含多个具有默认值的字段。但在返回内容中,只包含了明确设置字段值的字段nameprice

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from typing import List, Optional
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


# 定义数据模型
class Data(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: float = 10.5
tags: List[str] = []


datas = {
"min": {"name": "最小化", "price": 50.2},
"max": {"name": "最大化", "description": "都有值", "price": 62, "tax": 20.2},
"same": {"name": "默认", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}


# 定义路径参数
@app.get("/data/{data_id}", response_model=Data, response_model_exclude_unset=True)
async def read_item(data_id: str):
return datas[data_id]


if __name__ == '__main__':
uvicorn.run(app=app)

发送请求,示例代码:

1
2
3
curl -X 'GET' \
'http://127.0.0.1:8000/data/min' \
-H 'accept: application/json'

运行结果:

1
2
3
4
{
"name": "最小化",
"price": 50.2
}

除了response_model_exclude_unset,还可以在路径操作函数的装饰器中使用以下参数:

  1. response_model_exclude_defaults=True,忽略与默认值相同的字段。
  2. response_model_exclude_none=True,忽略值为None的字段。
  3. response_model_include={},输出数据中仅包含指定的字段。
  4. response_model_exclude={},输出数据中仅排除指定的字段。

这些参数都来自于Pydantic库,更多的内容可以参考Pydantic的官方文档中的模型导出部分。

使用多个响应模型

在有些场景下,一个路径操作函数需要返回不同的数据模型。
可以将Python中的类型联合体(Union)类设置为response_model对象值,在联合体中配置多个响应模型。

在下文的例子中,在路径操作函数的装饰器中,设置了响应模型为联合体,在联合体中设置了响应数据模型Dog和Cat。代码会根据逻辑,返回Dog或者Cat的响应模型。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
from typing import Union
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


# 定义数据模型基类,包含公用的字段
class BaseItem(BaseModel):
description: str
type: str


# 定义数据模型
class Cat(BaseItem):
# 定义字段type,默认值为cat
type = "cat"


# 定义数据模型
class Dog(BaseItem):
# 定义字段type,默认值为dog
type = "dog"
# 定义字段color
color: str


# 模拟数据,分别对应两个数据模型
items = {
"item1": {
"description": "三酷猫",
"type": "cat"
},
"item2": {
"description": "中华田园犬",
"type": "dog",
"color": "yellow",
},
}


# 使用Union返回多个数据模型
@app.get("/items/{item_id}", response_model=Union[Dog, Cat])
async def read_item(item_id: str):
return items[item_id]


if __name__ == '__main__':
uvicorn.run(app=app)

内置响应类

七种内置响应类

FastAPI中内置了以下响应类:

  1. 纯文本响应(PlainTextResponse)
  2. HTML响应(HTMLResponse)
  3. 重定向响应(RedirectResponse)
  4. JSON响应(JSONResponse)
  5. 通用响应(Response)
  6. 流响应(StreamingResponse)
  7. 文件响应(FileResponse)

纯文本响应

纯文本响应(PlainTextResponse),服务器端将一段纯文本写入响应后,直接返回给客户端,FastAPI不会对纯文本响应的内容做任何校验和转换。

使用纯文本响应的方法,在路径操作函数的装饰器中使用参数response_class指定响应类为PlainTextResponse类,然后FastAPI会将路径操作函数的返回值转换成字符串直接返回。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from fastapi import FastAPI
# 导入响应类
from fastapi.responses import PlainTextResponse
import uvicorn

app = FastAPI()


# 设置响应类为纯文本
@app.get("/", response_class=PlainTextResponse)
async def main():
return "Hello World"


if __name__ == '__main__':
uvicorn.run(app=app)

HTML响应

FastAPI内置了HTMLResponse类,用于处理HTML数据的响应返回。

在下文的例子中,用了两种方式返回HTML内容。

  1. 路径操作函数read_items1在装饰器中设置了参数response_class的值为HTMLResponse,浏览器会将返回的内容按照HTML格式解析。
  2. 路径操作函数read_items2将HTML内容文本写入HTMLResponse对象并返回响应,没有在装饰器中设置response_class。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
import uvicorn

app = FastAPI()

html_content = """
<html>
<head>
<title>浏览器顶部的标题</title>
</head>
<body>
<h1>三酷猫</h1>
</body>
</html>
"""


# 设置响应类为HTML响应
@app.get("/html/", response_class=HTMLResponse)
async def read_items1():
# 直接返回HTML文本
return html_content


# 未设置响应类
@app.get("/default/")
async def read_items2():
# 使用HTMLResponse对象返回数据
return HTMLResponse(content=html_content)


if __name__ == '__main__':
uvicorn.run(app=app)

重定向响应

在FastAPI中,内置的RedirectResponse类用于处理HTTP重定向的需求。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from fastapi import FastAPI
# 导入重定向类
from fastapi.responses import RedirectResponse
import uvicorn

app = FastAPI()


@app.get("/kaka")
async def read_three():
# 返回重定向响应
return RedirectResponse("https://kakawanyifan.com")


if __name__ == '__main__':
uvicorn.run(app=app)

JSON响应

在路径操作函数中,可以返回Response类或者任意Response的子类,比如JSONResponse。

默认情况下,FastAPI会使用jsonable_encoder()方法将模型数据转换成JSON格式,然后FastAPI会将这些JSON格式的数据放到一个JSONResponse类实例中,接着将该类的实例作为响应数据返回给客户端。

在下文的例子中,没有直接返回数据模型,而是使用jsonable_encoder()方法将数据模型转换成JSON格式的数据,再写入JSONResponse中并返回,这就是 FastAPI返回数据的过程。

在通常情况下,我们可以直接返回数据模型Item的实例item,不需要转换之后再返回。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from datetime import datetime
from typing import Optional
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import uvicorn


# 定义数据模型
class Item(BaseModel):
title: str
timestamp: datetime
description: Optional[str] = None


app = FastAPI()


# 注册路由路径
@app.post("/item/")
def update_item(item: Item):
# 使用jsonable_encoder转换数据模型
json_compatible_item_data = jsonable_encoder(item)
# 直接返回JSONResponse对象
return JSONResponse(content=json_compatible_item_data)


if __name__ == '__main__':
uvicorn.run(app=app)

通用响应

当返回数据不是JSON格式时,可以直接使用Response类响应对象。

在下文用Response类对象返回一段XML。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from fastapi import FastAPI, Response
from typing import Optional
import uvicorn

app = FastAPI()


# 注册路由路径路径
@app.get("/document/")
# 定义可选查询参数
def get_legacy_data(id: Optional[int] = None):
data = """<?xml version="1.0" encoding="utf-8" ?>
<Document>
<Header>
这里是页头
</Header>
<Body>
这里是内容
</Body>
<Footer>
这里是页脚
</Footer>
</Document>
"""
# 直接返回Response对象,并指定media_type="application/xml"
return Response(content=data, media_type="application/xml")


if __name__ == '__main__':
uvicorn.run(app=app)

访问http://127.0.0.1:8000/document/,响应如下:
XML

通过响应内容,我们可以看到,网页上显示的内容不是JSON格式,而是程序中指定的XML格式,并且FastAPI也没有对Response类对象的数据进行校验和转换。

流响应

在客户端和Web服务器端之间进行的数据传输,除了使用带有格式的文本数据以外,还可以使用字节流(Stream)进行传输。字节流响应的内容是二进制格式的,比如音频、视频、图片等。

在FastAPI中通过StreamingResponse类实现字节流响应。

在下文的例子中,使用open方法打开视频文件,得到文件流,接着使用StreamingResponse类的实例返回文件流的内容。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from fastapi import FastAPI
# 导入流响应
from fastapi.responses import StreamingResponse
import uvicorn

some_file_path = "large-video-file.mp4"
app = FastAPI()


# 注册路由路径
@app.get("/")
def main():
# 打开文件
file_like = open(some_file_path, mode="rb")
# 返回流并指定媒体类型
return StreamingResponse(file_like, media_type="video/mp4")


if __name__ == '__main__':
uvicorn.run(app=app)

文件响应

在FastAPI中,通过FileResponse类处理异步文件响应,与上一节的流响应相比,文件响应类在实例化时可以接收更多的参数。例如:

  1. path,要流式传输的文件的文件路径。
  2. headers,任何自定义响应头,传入字典类型。
  3. media_type,给出媒体类型的字符串;如果未设置,则文件名或路径将用于推断媒体类型。
  4. filename,如果给出,它将包含在响应Header的Content-Disposition中。

文件响应将包含Content-Length、Last-Modified和ETag的响应头,这些信息都将传递给浏览器,作为浏览器处理文件响应的依据。

在下文的例子中,导入文件响应类,并且在路径操作函数中将FileResponse类实例化,指定文件路径为参数,通过这种方式,就可以实现文件的异步下载功能。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from fastapi import FastAPI
# 导入文件响应
from fastapi.responses import FileResponse
import uvicorn

some_file_path = "large-video-file.mp4"
app = FastAPI()


# 注册路由路径
@app.get("/")
async def main():
# 直接使用文件名参数返回文件响应
return FileResponse(some_file_path)


if __name__ == '__main__':
uvicorn.run(app=app)

与使用StreamingResponse类不同。

  • 使用StreamingResponse类,需要先将文件打开,载入文件对象中进行返回,文件内容是一次性读取的,如果文件很大,就会占用大量的内存。
  • 使用FileResponse类,通过文件路径指定生成了一个FileResponse类实例,文件是异步读取的,会占用更少的内存。

所以,在实际的场景中,当需要直接处理流(Stream)时,使用StreamingResponse类; 当处理文件时,使用FileResponse类。

文章作者: Kaka Wan Yifan
文章链接: https://kakawanyifan.com/10905
版权声明: 本博客所有文章版权为文章作者所有,未经书面许可,任何机构和个人不得以任何形式转载、摘编或复制。

评论区