三万字长文让你彻底掌握 FastAPI
Posted Python猫
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了三万字长文让你彻底掌握 FastAPI相关的知识，希望对你有一定的参考价值。
楔子
随着 Python 的发展，与协程相关的 Web 框架也层出不穷，其中最受欢迎的莫过于 FastAPI。相比其它的协程框架，FastAPI 要更加的成熟，社区也更加的活跃。
那么 FastAPI 都有哪些特点呢？
快速：拥有非常高的性能，归功于 Starlette 和 Pydantic；Starlette 用于路由匹配，Pydantic 用于数据验证；
开发效率：功能开发效率提升 200% 到 300%；
减少 bug：减少 40% 的因为开发者粗心导致的错误；
智能：内部的类型注解非常完善，IDE 可处处自动补全；
简单：框架易于使用，文档易于阅读；
简短：使代码重复最小化，通过不同的参数声明实现丰富的功能；
健壮：可以编写出线上使用的代码，并且会自动生成交互式文档；
标准化：兼容 API 相关开放标准；
FastAPI 最大的特点就是它使用了 Python 的类型注解，我们后面会详细说，下面来安装一下 FastAPI。
使用 FastAPI 需要 Python 版本大于等于 3.6
安装很简单，直接 pip install fastapi 即可，并且会自动安装 Starlette 和 Pydantic。然后还要 pip install uvicorn，因为 uvicorn 是运行相关应用程序的服务器。或者一步到位：pip install fastapi[all]，会将所有依赖全部安装。
请求与响应
我们来使用 FastAPI 编写一个简单的应用程序：
from fastapi import FastAPI
import uvicorn
# 类似于 app = Flask(__name__)
app = FastAPI()
# 绑定路由和视图函数
@app.get("/")
async def index():
return "name": "古明地觉"
# 在 Windows 中必须加上 if __name__ == "__main__"
# 否则会抛出 RuntimeError: This event loop is already running
if __name__ == "__main__":
# 启动服务，因为我们这个文件叫做 main.py
# 所以需要启动 main.py 里面的 app
# 第一个参数 "main:app" 就表示这个含义
# 然后是 host 和 port 表示监听的 ip 和端口
uvicorn.run("main:app", host="0.0.0.0", port=5555)整个过程显然很简单，然后我们在浏览器中输入 localhost:5555 就会显示相应的输出。注意这里的视图函数，里面返回了一个字典，当然除了字典，其它的数据类型也是可以的，举个例子：
from fastapi import FastAPI
import uvicorn
app = FastAPI()
@app.get("/int")
async def index1():
return 666
@app.get("/str")
async def index2():
return "古明地觉"
@app.get("/bytes")
async def index3():
return b"satori"
@app.get("/tuple")
async def index4():
return ("古明地觉", "古明地恋", "雾雨魔理沙")
@app.get("/list")
async def index5():
return ["name": "古明地觉", "age": 17,
"name": "古明地恋", "age": 16]
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)这里我们直接使用 requests 发请求，测试一下。
可以看到基本上都是支持的，只不过元组自动转成列表返回了。并且当前的路径是写死的，如果我们想动态声明路径参数该怎么做呢？
from fastapi import FastAPI
import uvicorn
app = FastAPI()
@app.get("/items/item_id")
async def get_item(item_id):
"""
和 Flask 不同，Flask 是使用 <>
而 FastAPI 使用
"""
return "item_id": item_id
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)
整体非常简单，路由里面的路径参数可以放任意个，只是 里面的参数必须要在视图函数的参数中出现。但是问题来了，我们好像没有规定类型啊，如果我们希望某个路径参数只能接收指定的类型要怎么做呢？
from fastapi import FastAPI
import uvicorn
app = FastAPI()
@app.get("/items/item_id")
async def get_item(item_id: int):
"""
和 Flask 不同，Flask 定义类型是在路由当中
也就是在 <> 里面，变量和类型通过 : 分隔
而 FastAPI 是使用类型注解的方式
此时的 item_id 要求一个整型
准确的说是一个能够转成整型的字符串
"""
return "item_id": item_id
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)
如果我们传递的值无法转成整型的话，那么会进行提示：告诉我们 value 不是一个有效的整型，可以看到给的提示信息还是非常清晰的。
所以通过 Python 的类型声明，FastAPI 提供了数据校验的功能，当校验不通过的时候会清楚地指出没有通过的原因。在我们开发和调试的时候，这个功能非常有用。
交互式文档
FastAPI 会自动提供一个类似于 Swagger 的交互式文档，我们输入 localhost:5555/docs 即可进入。
注意一下左上角的 /openapi.json，可以点进去，会发现里面包含了我们定义的路由信息。
浏览器的话，由于我这里没有安装解析 JSON 的插件，所以看起来很不舒服。因此推荐大家安装一个 JSON Viewer 插件，查看 JSON 数据时会很方便。
至于 localhost:5555/docs 页面本身，我们也是可以进行设置的：
from fastapi import FastAPI
import uvicorn
app = FastAPI(title="测试文档",
description="这是一个简单的 demo",
docs_url="/my_docs",
openapi_url="/my_openapi")
@app.get("/items/item_id")
async def get_item(item_id: int):
return "item_id": item_id
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)然后我们再重新进入，此时在浏览器里就需要输入 localhost:5555/my_docs：
整体没什么难度，我们还可以指定其它参数，比如 version 等等，可以自己试试。该页面主要用来测试自己编写的 API 服务，不过个人更喜欢使用 requests 发请求。
路由顺序
我们在定义路由的时候需要注意一下顺序，举个例子：
from fastapi import FastAPI
import uvicorn
app = 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: int):
return "user_id": user_id
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)因为路径匹配是按照顺序进行的，所以这里要保证 /users/me 在 /users/user_id 的前面，否则的话只会匹配到 /users/user_id，这样的话访问 /users/me 就会解析错误，因为字符串 "me" 无法解析成整型。
使用枚举
我们可以将某个路径参数通过类型注解的方式声明为指定的类型（准确的说是可以转成指定的类型，因为默认都是字符串），但如果我们希望它只能是规定的几个值之一，该怎么做呢？
from enum import Enum
from fastapi import FastAPI
import uvicorn
app = FastAPI()
class Name(str, Enum):
satori = "古明地觉"
koishi = "古明地恋"
marisa = "雾雨魔理沙"
@app.get("/users/user_name")
async def get_user(user_name: Name):
return "user_id": user_name
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)通过枚举的方式可以实现这一点，我们来测试一下：
结果和我们期望的是一样的，可以再来看看 docs 生成的文档：
可用的值都有哪些，也自动提示给我们了。
路径中包含 /
假设我们有这样一个路由：/files/file_path，而用户传递的 file_path 中显然是可以带 / 的。假设 file_path 是 /root/test.py，那么路由就变成了 /files//root/test.py，显然这是有问题的。
那么为了防止解析出错，我们需要做一个类似于 Flask 的操作：
from fastapi import FastAPI
import uvicorn
app = FastAPI()
# 声明 file_path 的类型为 path
# 这样它会被当成一个整体
@app.get("/files/file_path:path")
async def get_file(file_path: str):
return "file_path": file_path
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)然后来访问一下：
结果没有问题，如果不将 file_path 的格式指定为 path，那么解析的时候就会找不到指定的路由。
查询参数
查询参数在 FastAPI 中依旧可以通过类型注解的方式进行声明，如果函数中定义了不属于路径参数的参数时，它们将会被解释为查询参数。
from fastapi import FastAPI
import uvicorn
app = FastAPI()
@app.get("/users/user_id")
async def get_user(user_id: str, name: str, age: int):
"""
我们在函数中定义了 user_id、name、age 三个参数
显然 user_id 和 路径参数中的 user_id 对应
然后 name 和 age 会被解释成查询参数
这三个参数的顺序没有要求，但一般都是路径参数在前，查询参数在后
"""
return "user_id": user_id, "name": name, "age": age
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)注意：name 和 age 没有默认值，这意味着它们是必须要传递的，否则报错。
我们看到当不传递 name 和 age 的时候，会直接提示你相关的错误信息。如果我们希望用户可以不传递的话，那么必须要指定一个默认值。
from fastapi import FastAPI
import uvicorn
app = FastAPI()
@app.get("/users/user_id")
async def get_user(user_id: str,
name: str = "UNKNOWN",
age: int = 0):
return "user_id": user_id, "name": name, "age": age
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)
这里使用了默认值，并且对于查询参数，由于它们指定了类型，所以我们也要传递正确类型的数据。假设给这里的 age 传递了一个 "abc"，那么是通不过的，因为它要求的是整型。
另外默认值的类型和指定的类型还可以不相同。
from fastapi import FastAPI
import uvicorn
app = FastAPI()
@app.get("/users/user_id")
async def get_user(user_id: str,
name: str = "UNKNOWN",
age: int = "蛤蛤蛤"):
return "user_id": user_id, "name": name, "age": age
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)这里的 age 需要接收一个整型，但默认值却是一个字符串，那么此时会有什么情况发生呢？我们来试一下：
我们看到，传递的 age 依旧需要整型，只不过在不传的时候会使用字符串类型的默认值。所以指定的类型和默认值类型不相同，也是允许的，只不过这么做显然是不合理的。
此外我们还可以指定多个类型，比如让 user_id 按照整型解析、解析不成功时退化为字符串。
from typing import Union, Optional
from fastapi import FastAPI
import uvicorn
app = FastAPI()
@app.get("/users/user_id")
async def get_user(user_id: Union[int, str],
name: Optional[str] = None):
"""
通过 Union 来声明一个混合类型，int 在前、str 在后
会先按照 int 解析，解析失败再变成 str
然后是 name，它表示字符串类型、但默认值为 None（不是字符串）
那么应该声明为 Optional[str]
"""
return "user_id": user_id, "name": name
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)
所以 FastAPI 的设计还是非常不错的，通过 Python 的类型注解来实现参数类型的限定可以说是非常巧妙的，因此这也需要我们熟练掌握 Python 的类型注解。
这里补充一下，我当前的 Python 版本是 3.8，如果你用的是 3.10，那么类型注解还有不同的写法：
>>> from typing import Union, Optional
# Optional[str] 和 str | None 等价
>>> name: Optional[str] = "古明地觉"
>>> name: str | None = "古明地觉"
# Union[int, str] 和 int | str 等价
>>> age: Union[int, str] = 17
>>> age: int | str = 17这种写法在 3.10 才开始正式引入，但通过 from __future__ import annotations 也可以在 3.9 里面使用，而 3.8 是不支持的。
布尔类型自动转换
对于布尔类型，FastAPI 支持自动转换，举个例子：
@app.get("/flag")
async def get_flag(flag: bool):
return "flag": flag
多个路径和查询参数
前面说过，可以定义任意个路径参数，只要动态的路径参数 在函数的参数中都出现即可。当然查询参数也可以是任意个，FastAPI 可以处理的很好。
@app.get("/postgres/schema/v1/table")
async def get_data(schema: str,
table: str,
select: str = "*",
where: Optional[str] = None,
limit: Optional[int] = None,
offset: Optional[int] = None):
"""
标准格式是：路径参数按照顺序在前，查询参数在后
但 FastAPI 对顺序本身是没有什么要求的
"""
query = f"select select from schema.table"
if where:
query += f" where where"
if limit:
query += f" limit limit"
if offset:
query += f" offset offset"
return "query": query然后使用 requests 测试一下：
Depends
这个老铁比较特殊，它是用来做什么的呢？我们来看一下：
from typing import Optional
from fastapi import FastAPI, Depends
import uvicorn
app = FastAPI()
async def common_parameters(
select: str = "*",
skip: int = 0,
limit: int = 100):
return "select": select, "skip": skip, "limit": limit
@app.get("/items/")
async def read_items(
commons: dict = Depends(common_parameters)):
# common_parameters 接收三个参数：select、skip、limit
# 因此会从请求中解析出 select、skip、limit 并传给 common_parameters
# 然后将 common_parameters 的返回值赋给 commons
# 但如果解析不到某个参数，那么会判断函数中参数是否有默认值
# 没有的话就会返回错误
return commons
@app.get("/users/")
async def read_users(
commons: dict = Depends(common_parameters)):
return commons
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)我们来测试一下：
所以 Depends 能够很好地实现依赖注入，而且这里特意写了两个路由，就是想表明它们是彼此独立的。因此当有共享的逻辑、或者共享的数据库连接、增强安全性、身份验证、角色权限等需求时，会非常的实用。
数据校验（针对查询参数）
FastAPI 支持我们进行更加智能的数据校验，比如一个字符串，我们希望用户在传递的时候只能传递长度为 6 到 15 的字符串该怎么做呢？
from typing import Optional
from fastapi import FastAPI, Query
import uvicorn
app = FastAPI()
@app.get("/user")
async def check_length(
# 默认值为 None，应该声明为 Optional[str]，当然声明 str 也是可以的
# 只不过声明为 str，那么默认值应该也是 str
# 所以如果允许一个类型的值为空，那么更规范的做法应该是声明为 Optional[类型]
password: Optional[str] = Query(None, min_length=6, max_length=15)
):
return "password": password
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)password 是可选的，但传递的时候必须传字符串、而且还是长度在 6 到 15 之间的字符串。所以在声明默认值的时候 None 和 Query(None) 是等价的，只不过 Query 还支持对参数进行额外的限制。
Query 里面除了限制最小长度和最大长度，还有其它的功能：
@app.get("/user")
async def check_length(
password: str = Query("satori", min_length=6,
max_length=15, regex=r"^satori")
):
"""
此时 password 的默认值为 'satori'，并且传递的时候也必须要以 'satori' 开头
但值得注意的是 password 后面的类型注解是 str，不再是 Optional[str]
因为默认值不是 None 了，当然这里即使写成 Optional[str] 也是没有什么影响的
"""
return "password": password
声明查询参数为必传参数
如果我们想让某个查询参数为必传参数，只需要不给它默认值就行了。
@app.get("/user")
async def check_length(password: str):
return "password": password函数里面的参数，要么是路径参数、要么是查询参数。显然 password 是一个查询参数，通过不指定默认值，我们即可实现它是一个必传参数。也就是在 URL 中，必须通过 ?password=xxx 的方式进行传递。
虽然目的很简单，但我们发现此时无法对 password 进行限制了，比如希望它的长度是 6 到 15。那么问题来了，如何才能两者兼顾呢？
@app.get("/user")
async def check_length(
password: str = Query(..., min_length=6,
max_length=15)
):
# 我们知道 Query 的第一个参数是 password 的默认值
# 但如果将 Query 的第一个参数换成 ...
# 那么 FastAPI 就不会再将它当成是默认值了
# 而是对 password 起一个限定作用，表示它是必传参数
return "password": password
... 是 Python 的一个特殊的对象，可以了解一下，在 Numpy 里面也可以使用它。
最后再补充一点，我们也可以不使用 Query，将 password 的长度限制逻辑写在函数体里面也是一样的。
同时获取多个相同的查询参数
如果我们指定了 a=1&a=2，那么在获取 a 的时候，会得到什么呢？
from typing import List
from fastapi import FastAPI, Query
import uvicorn
app = FastAPI()
@app.get("/items")
async def read_items(
a1: str = Query(...),
a2: List[str] = Query(...),
b: List[str] = Query(...)
):
return "a1": a1, "a2": a2, "b": b我们访问一下，看看结果：
首先 a2 和 b 都是列表，会获取所有的值，但 a1 只获取了最后一个值。
另外可能有人觉得代码有点啰嗦，在函数声明中可不可以这样写呢？
@app.get("/items")
async def read_items(
a1: str,
a2: List[str],
b: List[str]
):
return "a1": a1, "a2": a2, "b": b我们将 Query(...) 去掉了，因为它没有对参数做其它的限制，只是表示参数是一个必传参数。而不指定 Query(...)，那么本身就是一个必传参数，所以完全可以把 Query(...) 去掉。
这种做法，对于 a1 来说是可行的，但对于 a2 和 b 来说不行。对于类型为 list 的查询参数，我们必须要显式的加上 Query(...) 来表示必传参数。如果允许为 None（或者有默认值）的话，那么应该这么写：
@app.get("/items")
async def read_items(
a1: str,
a2: Optional[List[str]] = Query(None),
b: List[str] = Query(["1", "嘿嘿"])
):
return "a1": a1, "a2": a2, "b": b
给参数起别名
问题来了，假设我们定义的查询参数名叫 item-query，那么由于它要体现在函数参数中，而这显然不符合 Python 变量的命名规范，这个时候要怎么做呢？
@app.get("/items")
async def read_items(
# 三个查询参数，分别是 item-query、@@@@、$$$$
# 但它们不符合 Python 变量的命名规范
# 于是要为它们起别名
item1: Optional[str] = Query(None, alias="item-query"),
item2: str = Query("哈哈", alias="@@@@"),
# item3 是必传的
item3: str = Query(..., alias="$$$$")
):
return "item-query": item1, "@@@@": item2, "$$$$": item3
数值检测
Query 不仅仅支持对字符串的校验，还支持对数值的校验，里面可以传递 gt, ge, lt, le 等参数，相信这几个参数不用说你也知道是干什么的，我们举例说明：
@app.get("/items")
async def read_items(
# item1 必须大于 5
item1: int = Query(..., gt=5),
# item2 必须小于等于 7
item2: int = Query(..., le=7),
# item3 必须等于 10
item3: int = Query(..., ge=10, le=10)
):
return "item1": item1,
"item2": item2,
"item3": item3
Query 还是比较强大的 ，当然内部还有一些其它的参数是针对 docs 交互文档的，有兴趣可以自己了解一下。
数据校验（针对路径参数）
对查询参数进行数据校验使用的是 Query，对路径参数进行数据校验使用的是 Path，两者的使用方式一模一样，没有任何区别。
from fastapi import FastAPI, Path
import uvicorn
app = FastAPI()
@app.get("/items/item-id")
async def read_items(
item_id: int = Path(..., alias="item-id")
):
return "item-id": item_id路径参数是必须的，它是路径的一部分，所以我们应该使用 ... 将其标记为必传参数。当然即使不标记也无所谓，就算指定了默认值也用不上，因为路径参数不指定压根就匹配不到相应的路由。至于一些其它的校验，和查询参数一模一样，所以这里不再赘述了。
不过我们之前说过，路径参数应该在查询参数的前面，尽管 FastAPI 没有这个要求，但是这样写明显更舒服一些。不过问题来了，如果路径参数需要指定别名，但是某一个查询参数不需要，这个时候就会出现问题：
@app.get("/items/item-id")
async def read_items(
q: str,
item_id: int = Path(..., alias="item-id")
):
return "item_id": item_id, "q": q显然此时 Python 的语法就决定了 item_id 必须放在 q 的后面，当然这么做是完全没有问题的，FastAPI 对参数的先后顺序没有任何要求，因为它是通过参数的名称、类型和默认值声明来检测参数，而不在乎参数的顺序。但如果我们就要让 item_id 在 q 的前面要怎么做呢？
@app.get("/items/item-id")
async def read_items(
*,
item_id: int = Path(..., alias="item-id"),
q: str,
):
return "item_id": item_id, "q": q此时就没有问题了，通过将第一个参数设置为 *，使得 item_id 和 q 都必须通过关键字参数传递，所以此时默认参数在非默认参数之前也是允许的。当然我们也不需要担心 FastAPI 传参的问题，你可以认为它所有的参数都是通过关键字参数的方式传递的。
请求的载体：Request 对象
任何一个请求都对应一个 Request 对象，请求的所有信息都在这个 Request 对象中，FastAPI 也不例外。
from fastapi import FastAPI, Request
import uvicorn
app = FastAPI()
@app.get("/girl/user_id")
async def read_info(user_id: str,
request: Request):
"""
路径参数必须要体现在函数参数中
但是查询参数可以不写了
因为我们定义了 request: Request
那么请求相关的所有信息都会进入到这个 Request 对象中
"""
header = request.headers # 请求头
method = request.method # 请求方法
cookies = request.cookies # cookies
query_params = request.query_params # 查询参数
return "name": query_params.get("name"),
"age": query_params.get("age"),
"hobby": query_params.getlist("hobby")
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)通过 Request 对象可以获取请求相关的所有信息，我们之前参数传递不对的时候，FastAPI 会自动帮我们返回错误信息。但通过 Request 我们就可以自己进行解析、自己指定返回的错误信息了。
FastAPI 重度依赖 Python 的类型注解，假设 request 参数的类型是 str，那么 FastAPI 就会认为 request 是一个普通的查询参数。但这里 request 的类型是 Request，那么 FastAPI 就知道它代表整个请求，于是会自动将请求的载体 Request 对象赋值给参数 request。
而通过 request，我们可以拿到所有的请求参数，非常方便。只是数据校验这一步就必须由我们手动做了，比如这里 name 没有做校验，客户端传递任何值都是合法的，并且不传递的话也会返回 None。但手动校验的好处就是自由程度要更高一些，当参数不合法时，我们可以自定制返回的错误信息，之前的错误信息都是 FastAPI 内部预定义好的。
响应的载体：Response 对象
既然有 Request，那么必然会有 Response，虽然我们之前都是直接返回一个字典，但 FastAPI 实际上会帮我们转成一个 Response 对象。
Response 内部接收如下参数：
content：返回的数据；
status_code：状态码；
headers：返回的响应头；
media_type：响应类型（就是响应头里面的 Content-Type，这里单独作为一个参数出现了，其实通过 headers 参数设置也是可以的）；
background：接收一个任务，Response 在返回之后会自动异步执行（这里先不做介绍，后面会说）；
举个例子：
from fastapi import FastAPI, Request, Response
import uvicorn
import orjson
app = FastAPI()
@app.get("/girl/user_id")
async def read_info(user_id: str,
request: Request):
# 查询参数
query_params = request.query_params
data = "name": query_params.get("name"),
"age": query_params.get("age"),
"hobby": query_params.getlist("hobby")
# 实例化一个 Response 对象
response = Response(
# content，手动转成 json
orjson.dumps(data),
# status_code，状态码
201,
# headers，响应头
"Token": "xxx",
# media_type，就是 html 中的 Content-Type
# content 只是一坨字节流，需要告诉客户端响应类型
# 这样客户端才能正确的解析
"application/json",
)
# 拿到 response 的时候，还可以单独对响应头和 cookie进行设置
response.headers["ping"] = "pong"
# 设置 cookie 的话，通过 response.set_cookie
response.set_cookie("SessionID", "abc123456")
# 也可以通过 response.delete_cookie 删除 cookie
return response
通过 Response 我们可以实现请求头、状态码、cookie 的自定义。另外除了 Response 之外还有很多其它类型的响应，比如：
FileResponse：用于返回文件；
HTMLResponse：用于返回 HTML；
PlainTextResponse：用于返回纯文本；
JSONResponse：用于返回 JSON；
RedirectResponse：用于重定向；
StreamingResponse：用于返回二进制流；
它们都继承了 Response，只不过会自动帮你设置响应类型，举个例子：
from fastapi import FastAPI
from fastapi.responses import Response, HTMLResponse
import uvicorn
app = FastAPI()
@app.get("/index")
async def index():
response1 = HTMLResponse("<h1>你好呀</h1>")
response2 = Response("<h1>你好呀</h1>",
media_type="text/html")
# 以上两者是等价的，在 HTMLResponse 里面
# 会自动将 media_type 设置成 text/html
return response1另外我们在开头说过，FastAPI 的请求与响应相关的功能，实际上是基于 starlette。
请求载体 Request 和响应载体 Response 都是直接从 starlette 里面导入的。
其它类型的请求与响应
FastAPI 除了 GET 请求之外，还支持其它类型，比如：POST, PUT, DELETE, OPTIONS, HEAD, PATCH, TRACE 等等。而常见的也就 GET, POST, PUT, DELETE，介绍完了 GET，我们来说一说其它类型的请求。
显然对于 POST、PUT 等类型的请求，我们必须要能够解析出请求体。
Model
在 FastAPI 中，请求体可以看成是 Model 对象，举个例子：
from typing import Optional
from fastapi import FastAPI, Response
from pydantic import BaseModel
import orjson
import uvicorn
app = FastAPI()
class Girl(BaseModel):
"""
数据验证是通过 pydantic 实现的
我们需要从中导入 BaseModel，然后继承它
"""
name: str
age: Optional[str] = None
length: float
@app.post("/girl")
async def read_info(girl: Girl):
# girl 就是我们接收的请求体，它需要通过 json 来传递
# 并且这个 json 要有上面的三个字段（age 可以没有）
# 通过 girl.xxx 的方式我们可以获取和修改内部的所有属性
data = "姓名": girl.name, "年龄": girl.age,
"身高": girl.length
return Response(
orjson.dumps(data),
media_type="application/json"
)
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)我们访问一下：
除了使用 pydantic，我们还可以手动验证：
@app.post("/girl")
async def read_info(request: Request):
# 是一个协程，所以需要 await
data = await request.body()我们说过 Request 对象是请求的载体，它包含了请求的所有信息，代码中的 data 便是请求体，并且是最原始的字节流形式。而它长什么样子呢？
首先在使用 requests 模块发送 post 请求的时候，数据可以通过 data 参数传递、也可以通过 json 参数传输。
所以 await request.body()得到的就是最原始的字节流，除了它之外还有 await request.json()，它在内部依旧会获取字节流，只不过获取之后会自动 loads 成字典。
因此使用 await request.json() 也侧面要求，我们在发送请求的时候必须使用 json 参数传递，否则无法正确解析。
@app.post("/girl")
async def read_info(request: Request):
data = await request.body()
try:
# 解析成字典
data = orjson.loads(data)
except orjson.JSONDecodeError:
result = "error": "请传递 JSON"
return Response(
orjson.dumps(result),
status_code=404,
media_type="application/json"
)
result = "name": data.get("name"),
"age": data.get("age"),
"length": data.get("length")
return Response(
orjson.dumps(result),
media_type="Application/json"
)
从 Request 对象解析出请求体之后，我们手动转成了字典，如果你对字段有要求的话，那么可以再单独进行判断。
就我个人而言，基本很少使用 pydantic 做数据验证，一般都是手动解析数据、进行验证。当数据不合法时，返回自定义的错误信息。
路径参数、查询参数、请求体
这几种不同的参数，我们可以混合在一起：
from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn
app = FastAPI()
class Girl(BaseModel):
name: str
age: Optional[str] = None
length: float
@app.post("/girl/user_id")
async def read_info(user_id,
q: str,
girl: Girl):
# user_id：路径参数，q：查询参数，girl：请求体
return "user_id": user_id,
"q": q,
**dict(girl)
里面同时指定了路径参数、查询参数和请求体，FastAPI 依然是可以正确区分的，当然我们也可以使用 Request 对象。
@app.post("/girl/user_id")
async def read_info(user_id,
request: Request):
# user_id 是路径参数，它一定要出现在视图函数中
# 并且没有限制类型，那么 user_id 可以是任意类型
# 然后查询参数和请求体，可以通过 request 获取
q = request.query_params.get("q")
# 请求体应该是一个 JSON
data = await request.json()
return "user_id": user_id, "q": q, **data发请求的话，返回的内容是一样的。
所以对于服务端而言，解析有两种方式。一种是体现在函数参数中，如果参数不对，FastAPI 会自动检测到，然后抛出预定义错误；而另一种则是使用 Request 对象，此时请求相关的全部信息都会被封装到这个对象中，然后我们手动解析，当参数不合法时，可以自定义返回的错误信息，可控性更高。
特别是当 JSON 的字段非常多的时候，定义 Model 比较麻烦，用 Request 对象会方便一些。举个例子：
如果发送的 JSON 里面有很多字段，每个字段的值的类型还不同，以及还包含 JSON 的嵌套等等。那么再通过定义 Model 的方式就很麻烦了，而通过 Request 拿到字节流之后再解析，就会方便很多。
Form 表单
我们调用 requests.post，如果参数通过 data 传递的话，则相当于提交了一个 form 表单，那么在 FastAPI 中可以通过 await request.form() 进行获取，注意：内部同样会先调用 await request.body()。
@app.post("/girl")
async def read_info(request: Request):
form = await request.form()
return "name": form.get("name"),
"age": form.getlist("age")
而对于表单提交，FastAPI 还提供了另一种方式。
from fastapi import FastAPI, Form
import uvicorn
app = FastAPI()
@app.post("/user")
async def get_user(username: str = Form(...),
password: str = Form(...)):
return "username": username,
"password": password
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)
像 Form 表单，查询参数、路径参数等等，都可以和 Request 对象一起使用，像上面的例子。如果再多定义一个参数 request: Request，那么仍然可以通过 await request.form() 拿到相关的表单信息。
@app.post("/user")
async def get_user(*,
username: str = Form(...),
password: str = Form(...),
request: Request):
form = await request.form()
return "username": username,
"password": password
# 两个 return 是等价的
return "username": form.get("username"),
"password": form.get("password")所以如果你觉得某个参数不适合类型注解，那么可以单独通过 Request 对象进行解析，因为它是请求的载体，请求相关的一切信息都在里面。
文件上传
然后是文件上传功能，FastAPI 如何接收用户的文件上传呢？首先如果想支持文件上传，必须要安装一个包 python-multipart，直接用 pip 安装即可。
from fastapi import FastAPI, File, UploadFile
import uvicorn
app = FastAPI()
@app.post("/file1")
async def file1(file: bytes = File(...)):
# 此时会以字节流的形式拿到文件的具体内容
return "文件长度": len(file)
@app.post("/file2")
async def file2(file: UploadFile = File(...)):
# 会拿到文件句柄
# 通过 await file.read() 可拿到文件内容
return "文件名": file.filename,
"文件长度": len(await file.read())
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)
所以我们可以直接获取字节流，或者获取文件句柄。但如果是多个文件上传要怎么做呢？
from typing import List
from fastapi import FastAPI, UploadFile, File
import uvicorn
app = FastAPI()
@app.post("/file")
async def file(files: List[UploadFile] = File(...)):
"""
指定类型为列表即可
"""
return ["文件名": f.filename,
"文件长度": len(await f.read())
for f in files]
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)
此时我们就实现了 FastAPI 文件上传，当然文件上传并不影响我们处理表单，可以自己试一下同时处理文件和表单。
返回静态资源
再来看看 FastAPI 如何返回静态资源，首先我们需要安装 aiofiles，直接 pip 安装即可。
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
import uvicorn
app = FastAPI()
# name 参数只是起一个名字，FastAPI 内部使用
app.mount("/static",
StaticFiles(directory=r"/Users/satori/Downloads/pics"),
name="static")
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)浏览器输入：localhost:5555/static/1.png，那么会返回指定目录下的 1.png 文件。
APIRouter
APIRouter 类似于 Flask 的蓝图，可以更好地组织大型项目，举个例子：
在当前的工程目录中有一个 app 目录和一个 main.py，其中 app 目录里面有一个 app01.py，然后来看看它们是如何组织的。
# app/app01.py
from fastapi import APIRouter
router = APIRouter(prefix="/router")
# 以后访问的时候要通过 /router/v1 来访问
@router.get("/v1")
async def v1():
return "message": "hello world"
# main.py
from fastapi import FastAPI
from app.app01 import router
import uvicorn
app = FastAPI()
# 将 router 注册到 app 中
# 相当于 Flask 里面的 register_blueprint
app.include_router(router)
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)然后在外界便可以通过 /router/v1 的方式来访问。
错误处理
错误处理也是一个不可忽视的点，错误有很多种，比如：
客户端没有足够的权限执行此操作；
客户端没有访问某个资源的权限；
客户端尝试访问一个不存在的资源；
······
这个时候我们应该将错误通知给相应的客户端，这个客户端可以浏览器、代码程序、IoT 设备等等。
但是就我个人而言，更倾向于使用 Response 对象，将里面的 status_code 设置为 404，然后在返回的 json 中指定错误信息。不过 FastAPI 内部也提供了一些异常类：
from fastapi import FastAPI, HTTPException
import uvicorn
app = FastAPI()
@app.get("/items/item_id")
async def read_item(item_id: str):
if item_id != "foo":
# 里面还可以传入 headers 设置响应头
raise HTTPException(status_code=404,
detail="item 没有发现")
return "item": "bar"
HTTPException 是一个普通的 Python 异常类（继承了 Exception），它携带了 API 的相关信息。并且既然是异常，那么我们不能 return、而是要 raise。
个人觉得这个不是很常用，至少我本人很少用这种方式返回错误，因为它能够携带的信息太少了。
自定义异常
上面使用的 HTTPException 是 FastAPI 内部提供的异常类，我们也可以自定义，但是定义完异常之后，还要定义一个 handler，将异常和 handler 绑定在一起，然后引发该异常的时候就会触发相应的 handler。
from fastapi import FastAPI, Request
from fastapi.responses import Response
import uvicorn
import orjson
app = FastAPI()
class ASCIIException(Exception):
"""什么也不做"""
# 通过装饰器的方式
# 将 ASCIIException 和 ascii_exception_handler 绑定在一起
@app.exception_handler(ASCIIException)
async def ascii_exception_handler(request: Request,
exc: ASCIIException):
"""
当引发 ASCIIException 的时候，
会触发 ascii_exception_handler 的执行
同时会将 request 和 exception 传过去
"""
return Response(
orjson.dumps("code": 404,
"message": "必须传递 ascii 字符串"),
status_code=404
)
@app.get("/items/item_id")
async def read_item(item_id: str):
if not item_id.isascii():
raise ASCIIException
return "item": f"get item_id"
还是很简单的，另外关于 Request 和 Response，我们除了可以通过 fastapi 导入，还可以通过 starlette 导入，因为 fastapi 的路由映射是通过 starlette 来实现的。当然我们直接从 fastapi 里面导入即可。
自定义 404
当访问一个不存在的 URL，我们应该提示用户，比如：您要找到页面去火星了。
from fastapi import FastAPI
from fastapi.responses import Response
from fastapi.exceptions import StarletteHTTPException
import uvicorn
import orjson
app = FastAPI()
@app.exception_handler(StarletteHTTPException)
async def not_found(request, exc):
return Response(
orjson.dumps(
"code": 404,
"message": f"您要找的页面 request.url 去火星了。。。"),
status_code=404
)
此时访问一个不存在的 URL 时，就会返回我们自定义的 JSON 字符串。而参数 request，就是请求对应的 Request 对象，为了方便 IDE 提示，定义的时候可以加上一个类型注解。
后台任务
如果处理请求的时候需要执行一个耗时任务，那么可以将其放在后台执行，而 FastAPI 已经帮我们做好了这一步。来看一下：
import time
from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import Response
import uvicorn
import orjson
app = FastAPI()
def send_email(email: str, message: str = ""):
"""发送邮件，假设耗时三秒"""
time.sleep(3)
print(f"三秒之后邮件发送给 email!r, "
f"邮件信息: message!r")
@app.get("/user/email")
async def order(email: str, bg_tasks: BackgroundTasks):
"""
这里需要多定义一个参数
此时任务就被添加到后台，当 Response 对象返回之后触发
"""
# 可以添加任意多个任务
bg_tasks.add_task(send_email, email, message="这是一封邮件")
return Response(
orjson.dumps("message": "邮件发送成功")
)
# 我们在之前介绍 Response 的时候说过，里面有一个参数 background
# 所以我们还可以这么做
"""
bg_tasks = BackgroundTasks() # 不在参数中定义 bg_tasks
bg_tasks.add_task(send_email, email, message="这是一封邮件")
return Response(
orjson.dumps("message": "邮件发送成功"),
background=bg_tasks
)
"""
以上是关于三万字长文让你彻底掌握 FastAPI的主要内容，如果未能解决你的问题，请参考以下文章