月薪40k+测试·开发同步认可的FastAPI：Python 世界里最受欢迎的异步框架

目录

介绍：

这次我们来聊一聊 FastAPI，它和我们之前介绍的 Sanic 是类似的，都是 Python 中的异步 web 框架。相比 Sanic，FastAPI 更加的成熟、社区也更加的活跃，那么 FastAPI 都有哪些特点呢？

请求与响应

交互式文档

路由顺序

使用枚举

路径中包含 /

查询参数

查询参数和数据校验

路径参数和数据校验

Request

Response

其它类型的请求与响应

返回静态资源

错误处理

自定义异常

自定义 404

后台任务

APIRouter

中间件

高阶操作

其它的响应

HTTP 验证

websocket

FastAPI 服务的部署

小结

兄弟们如果想要北凡的完整的精品教学视频记得加微信哈，还有笔记哦，有什么不懂可以问蓓蓓小姐姐或者北凡老师哦

学习资料可以找到我们呢的蓓蓓小姐姐【mashang-qq】备注【csdn000】免费领取哦

介绍：

这次我们来聊一聊 FastAPI，它和我们之前介绍的 Sanic 是类似的，都是 Python 中的异步 web 框架。相比 Sanic，FastAPI 更加的成熟、社区也更加的活跃，那么 FastAPI 都有哪些特点呢？

FastAPI 最大的特点就是它使用了 Python 的类型注解，我们后面会详细说，下面来安装一下 FastAPI。

使用 FastAPI 需要 Python 版本大于等于 3.6。

首先是 pip install fastapi，会自动安装 Starlette 和 Pydantic；然后还要 pip install uvicorn，因为 uvicorn 是运行相关应用程序的服务器。或者一步到胃：pip install fastapi[all]，会将所有依赖全部安装。

请求与响应

我们来使用 FastAPI 编写一个简单的应用程序：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIimportuvicorn# 类似于 app = Flask(__name__)app = FastAPI()# 绑定路由和视图函数@app.get("/")asyncdefindex():return{"name": "古明地觉"}# 在 Windows 中必须加上 if __name__ == "__main__"，否则会抛出 RuntimeError: This event loop is already runningif__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" 就会显示相应的输出，我们看到在视图函数中可以直接返回一个字典。当然除了字典，其它的数据类型也是可以的，举个栗子：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIimportuvicornapp = FastAPI()@app.get("/int")asyncdefindex1():return666@app.get("/str")asyncdefindex2():return"古明地觉"@app.get("/bytes")asyncdefindex3():returnb"satori"@app.get("/tuple")asyncdefindex4():return("古明地觉", "古明地恋", "雾雨魔理沙")@app.get("/list")asyncdefindex5():return[{"name": "古明地觉", "age":17}, {"name": "古明地恋", "age":16}]if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

这里我们直接使用 requests 发请求：

我们看到基本上都是支持的，只不过元组自动转成列表返回了。这里我们在路由中指定了路径，可以看到 FastAPI 中的路径形式和其它框架并无二致，只不过目前的路径是写死的，如果我们想动态声明路径参数该怎么做呢？

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIimportuvicornapp = FastAPI()@app.get("/items/{item_id}")asyncdefget_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)

整体非常简单，路由里面的路径参数我们可以放任意个，只是{}里面的参数必须要在定义的视图函数的参数中出现。但是问题来了，我们好像没有规定类型啊，如果我们希望某个路径参数只能接收指定的类型要怎么做呢？

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIimportuvicornapp = FastAPI()@app.get("/apple/{item_id}")asyncdefget_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，我们可以点击进去，会发现里面包含了我们定义的路由信息。

至于"localhost:5555/docs" 页面本身，我们也是可以进行设置的：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIimportuvicornapp = FastAPI(title="测试文档",description="这是一个简单的 demo",docs_url="/my_docs",openapi_url="/my_openapi")@app.get("/apple/{item_id}")asyncdefget_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)

然后我们再重新进入，此时就需要通过"localhost:5555/my_docs"：

整体没什么难度，我们还可以指定其它参数，比如 version 表示版本，可以自己试试。该页面主要用来测试自己编写的 API 服务，不过个人更喜欢使用 requests 发请求。

路由顺序

然后我们在定义路由的时候需要注意一下顺序，举个栗子：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIimportuvicornapp = FastAPI()@app.get("/users/me")asyncdefread_user_me():return{"user_id": "the current user"}@app.get("/users/{user_id}")asyncdefread_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" 无法解析成整型。

使用枚举

我们可以将某个路径参数通过类型注解的方式声明为指定的类型（准确的说是可以转成指定的类型，因为默认都是字符串），但如果我们希望它只能是我们规定的几个值之一该怎么做呢？

# -*- coding:utf-8 -*-# @Author: komeiji satorifromenumimportEnumfromfastapiimportFastAPIimportuvicornapp = FastAPI()className(str, Enum):satori = "古明地觉"koishi = "古明地恋"marisa = "雾雨魔理沙"@app.get("/users/{user_name}")asyncdefget_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 中的操作：

fromfastapiimportFastAPIimportuvicornapp = FastAPI()# 声明 file_path 的类型为 path，这样它会被当成一个整体@app.get("/files/{file_path:path}")asyncdefget_file(file_path:str):return{"file_path": file_path}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后我们来访问一下：

查询参数

查询参数在 FastAPI 中依旧可以通过类型注解的方式进行声明，如果函数中定义了不属于路径参数的参数时，那么它们将会被自动解释会查询参数。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIimportuvicornapp = FastAPI()@app.get("/user/{user_id}")asyncdefget_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 的时候，会直接提示你相关的错误信息。如果我们希望用户可以不传递的话，那么必须要指定一个默认值。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIimportuvicornapp = FastAPI()@app.get("/user/{user_id}")asyncdefget_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"，那么也是通不过的。

如果默认值和类型不相同怎么办？

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIimportuvicornapp = FastAPI()@app.get("/user/{user_id}")asyncdefget_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 按照整型解析、解析不成功退化为字符串。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportUnion, OptionalfromfastapiimportFastAPIimportuvicornapp = FastAPI()@app.get("/user/{user_id}")asyncdefget_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 的类型注解。

bool 类型自动转换

对于布尔类型，FastAPI 支持自动转换，举个栗子：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIimportuvicornapp = FastAPI()@app.get("/{flag}")asyncdefget_flag(flag:bool):return{"flag": flag}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

多个路径和查询参数

我们之前说过，可以定义任意个路径参数，只要动态的路径参数{}里面的在函数的参数中都出现即可。当然查询参数也可以是任意个，FastAPI 可以处理的很好。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptionalfromfastapiimportFastAPIimportuvicornapp = FastAPI()@app.get("/postgres/{schema}/v1/{table}")asyncdefget_data(schema:str,table:str,select:str= "*",where: Optional[str] =None,limit: Optional[int] =None,offset: Optional[int] =None):"""标准格式是：路径参数按照顺序在前，查询参数在后但其实对顺序是没有什么要求的"""query = f"select {select} from {schema}.{table}"ifwhere:query += f" where {where}"iflimit:query += f" limit {limit}"ifoffset:query += f" offset {offset}"return{"query": query}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后我们使用 requests 来测试一下：

print(requests.get("http://localhost:5555/postgres/ods/v1/staff").json()) # {'query': 'select * from ods.staff'}print(requests.get("http://localhost:5555/postgres/ods/v1/staff?select=id, name&where=id > 3&limit=100").json()) # {'query': 'select id, name from ods.staff where id > 3 limit 100'}

Depends

这个老铁比较特殊，它是用来做什么的呢？我们来看一下：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptionalimportuvicornfromfastapiimportDepends, FastAPIapp = FastAPI()asyncdefcommon_parameters(q: Optional[str] =None, skip:int=0, limit:int=100):return{"q": q, "skip": skip, "limit": limit}@app.get("/items/")asyncdefread_items(commons:dict= Depends(common_parameters)):# common_parameters 接收三个参数：q、skip、limit# 然后在解析请求的时候，会将 q、skip、limit 传递到 common_parameters 中，然后将返回值赋值给 commons# 但如果解析不到某个参数时，那么会判断函数中参数是否有默认值，没有的话就会返回错误，而不是传递一个 None 进去returncommons@app.get("/users/")asyncdefread_users(commons:dict= Depends(common_parameters)):returncommonsif__name__ == "__main__":uvicorn.run("main1:app", host="0.0.0.0", port=5555)

我们来测试一下：

所以 Depends 能够很好的实现依赖注入，而且我们特意写了两个路由，就是想表示它们是彼此独立的。因此当有共享的逻辑、或者共享的数据库连接、增强安全性、身份验证、角色权限等等，会非常的实用。

查询参数和数据校验

FastAPI 支持我们进行更加智能的数据校验，比如一个字符串，我们希望用户在传递的时候只能传递长度为 6 到 15 的字符串该怎么做呢？

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptionalfromfastapiimportFastAPI, Queryimportuvicornapp = FastAPI()@app.get("/user")asyncdefcheck_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，那么在声明默认值的时候 None 和 Query(None) 是等价的，只不过 Query 还支持其它的参数来对参数进行限制。

Query 里面除了限制最小长度和最大长度，还有其它的功能：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, Queryimportuvicornapp = FastAPI()@app.get("/user")asyncdefcheck_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}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

声明为必须参数

我们通过 Query 可以限制参数的长度，但是问题来了，这个时候我还希望这个参数是必传的该怎么做呢？

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, Queryimportuvicornapp = FastAPI()@app.get("/user")asyncdefcheck_length(password:str= Query(..., min_length=6)):"""将第一个参数换成 ... 即可实现该参数是必传参数"""return{"password": password}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

... 是 Python 中的一个特殊的对象，可以了解一下，通过它可以实现该参数是必传参数。

查询参数变成一个列表

如果我们指定了a=1&a=2，那么我们在获取 a 的时候如何才能得到一个列表呢？

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptional, ListfromfastapiimportFastAPI, Queryimportuvicornapp = FastAPI()@app.get("/items")asyncdefread_items(a1:str= Query(...),a2: List[str] = Query(...),b: List[str] = Query(...)):return{"a1": a1, "a2": a2, "b": b}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

我们访问一下，看看结果：

首先"a2" 和"b" 都是对应列表，然后"a1" 只获取了最后一个值。另外可能有人觉得我们这样有点啰嗦，在函数声明中可不可以这样写呢？

@app.get("/items")asyncdefread_items(a1:str,a2: List[str],b: List[str]):return{"a1": a1, "a2": a2, "b": b}

对于 a1 是可以的，但是 a2 和 b 不行。对于类型为 list 的查询参数，无论有没有默认值，你都必须要显式的加上 Query 来表示必传参数。如果允许为 None（或者有默认值）的话，那么应该这么写：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptional, ListfromfastapiimportFastAPI, Queryimportuvicornapp = FastAPI()@app.get("/items")asyncdefread_items(a1:str,a2: Optional[List[str]] = Query(None),b: List[str] = Query(["1", "嘿嘿"])):return{"a1": a1, "a2": a2, "b": b}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

给参数起别名

问题来了，假设我们定义的查询参数名叫 item-query，那么由于它要体现在函数参数中、而这显然不符合 Python 变量的命名规范，这个时候要怎么做呢？答案是起一个别名。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptional, ListfromfastapiimportFastAPI, Queryimportuvicornapp = FastAPI()@app.get("/items")asyncdefread_items(# 通过 url 的时候使用别名即可item1: Optional[str] = Query(None, alias="item-query"),item2:str= Query("哈哈", alias="@@@@"),item3:str= Query(..., alias="$$$$") # item3 是必传的):return{"item1": item1, "item2": item2, "item3": item3}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

数值检测

Query 不仅仅支持对字符串的校验，还支持对数值的校验，里面可以传递 gt、ge、lt、le 这几个参数，相信这几个参数不用说你也知道是干什么的，我们举例说明：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, Queryimportuvicornapp = FastAPI()@app.get("/items")asyncdefread_items(# item1 必须大于 5item1:int= Query(..., gt=5),# item2 必须小于等于 7item2:int= Query(..., le=7),# item3 必须必须等于 10item3:int= Query(..., ge=10, le=10)):return{"item1": item1, "item2": item2, "item3": item3}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

Query 还是比较强大了，当然内部还有一些其它的参数是针对 docs 交互文档的，有兴趣可以自己了解一下。

路径参数和数据校验

查询参数数据校验使用的是 Query，路径参数数据校验使用的是 Path，两者的使用方式一模一样，没有任何区别。

fromfastapiimportFastAPI, Pathimportuvicornapp = FastAPI()@app.get("/items/{item-id}")asyncdefread_items(item_id:int= Path(..., alias="item-id")):return{"item_id": item_id}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

因为路径参数是必须的，它是路径的一部分，所以我们应该使用 ... 将其标记为必传参数。当然即使不这么做也无所谓，因为指定了默认值也用不上，因为路径参数不指定压根就匹配不到相应的路由。至于一些其它的校验，和查询参数一模一样，所以这里不再赘述了。

不过我们之前说过，路径参数应该在查询参数的前面，尽管 FastAPI 没有这个要求，但是这样写明显更舒服一些。但是问题来了，如果路径参数需要指定别名，但是某一个查询参数不需要，这个时候就会出现问题：

@app.get("/items/{item-id}")asyncdefread_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}")asyncdefread_items(*, item_id:int= Path(..., alias="item-id"),q:str):return{"item_id": item_id, "q": q}

此时就没有问题了，通过将第一个参数设置为 *，使得 item_id 和 q 都必须通过关键字传递，所以此时默认参数在非默认参数之前也是允许的。当然我们也不需要担心 FastAPI 传参的问题，你可以认为它所有的参数都是通过关键字参数的方式传递的。

Request

Request 是什么？首先我们知道任何一个请求都对应一个 Request 对象，请求的所有信息都在这个 Request 对象中，FastAPI 也不例外。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, Requestimportuvicornapp = FastAPI()@app.get("/girl/{user_id}")asyncdefread_girl(user_id:str,request: Request):"""路径参数是必须要体现在参数中，但是查询参数可以不写了因为我们定义了 request: Request，那么请求相关的所有信息都会进入到这个 Request 对象中"""header = request.headers # 请求头method = request.method # 请求方法cookies = request.cookies # cookiesquery_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 我们就可以自己进行解析、自己指定返回的错误信息了。

Response

既然有 Request，那么必然会有 Response，尽管我们可以直接返回一个字典，但 FastAPI 实际上会帮我们转成一个 Response 对象。

Response 内部接收如下参数：

content：返回的数据status_code：状态码headers：返回的请求头media_type：响应类型(就是 HTML 中 Content-Type，只不过这里换了个名字)background：接收一个任务，Response 在返回之后会自动异步执行（这里不做介绍，后面会说）

举个栗子：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, Request, Responseimportuvicornimportorjsonapp = FastAPI()@app.get("/girl/{user_id}")asyncdefread_girl(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 字符串，如果直接返回字典的话，那么在包装成 Response 对象的时候会自动帮你转orjson.dumps(data),# status_code，状态码201,# headers，响应头{"Token": "xxx"},# media_type，就是 HTML 中的 Content-Type"application/json", )# 如果想设置 cookie 的话，那么通过 response.set_cookie 即可# 删除 cookie 则是 response.delete_cookiereturnresponseif__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

通过 Response 我们可以实现请求头、状态码、cookie 等自定义。

另外除了 Response 之外还有很多其它类型的响应，它们都在 fastapi.responses 中，比如：FileResponse、HTMLResponse、PlainTextResponse 等等。它们都继承了 Response，只不过会自动帮你设置响应类型，举个栗子：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIfromfastapi.responsesimportResponse, HTMLResponseimportuvicornapp = FastAPI()@app.get("/index")asyncdefindex():response1 = HTMLResponse("<h1>你好呀</h1>")response2 = Response("<h1>你好呀</h1>", media_type="text/html")# 以上两者是等价的，在 HTMLResponse 中会自动将 media_type 设置成 text/htmlreturnresponse1if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

其它类型的请求与响应

FastAPI 除了 GET 请求之外，还支持其它类型，比如：POST、PUT、DELETE、OPTIONS、HEAD、PATCH、TRACE 等等。而常见的也就 GET、POST、PUT、DELETE，介绍完了 GET，我们来说一说其它类型的请求。

显然对应 POST、PUT 等类型的请求，我们必须要能够解析出请求体，并且能够构造出响应体。

Model

在 FastAPI 中，请求体和响应体都对应一个 Model，举个栗子：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptional, ListfromfastapiimportFastAPI, Request, ResponsefrompydanticimportBaseModelimportuvicornapp = FastAPI()classGirl(BaseModel):"""数据验证是通过 pydantic 实现的，我们需要从中导入 BaseModel，然后继承它"""name:strage: Optional[str] =Nonelength:floathobby: List[str] # 对于 Model 中的 List[str] 我们不需要指定 Query（准确的说是 Field）@app.post("/girl")asyncdefread_girl(girl: Girl):# girl 就是我们接收的请求体，它需要通过 json 来传递，并且这个 json 要有上面的四个字段（age 可以没有）# 通过 girl.xxx 的方式我们可以获取和修改内部的所有属性returndict(girl) # 直接返回 Model 对象也是可以的if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

除了使用这种方式之外，我们还可以使用之前说的 Request 对象：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, Requestimportuvicornapp = FastAPI()@app.post("/girl")asyncdefread_girl(request: Request):# 是一个协程，所以需要 awaitdata =awaitrequest.body()print(data)if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

首先我们在使用 requests 模块发送 post 请求的时候可以通过 data 参数传递、也可以通过 json 参数。

当通过json={"name": "satori", "age": 16, "length": 155.5}传递的时候，会将其转成 json 字符串进行传输，程序中的 print 打印如下：

b'{"name": "satori", "age": 16, "length": 155.5}'

如果我们是用过 data 参数发请求的话（值不变），那么会将其拼接成k1=v1&k2=v2的形式再进行传输（相当于表单提交，后面说），程序中打印如下：

b'name=satori&age=16&length=155.5'

所以我们看到 await request.body() 得到的就是最原始的字节流，而除了 await request.body() 之外还有一个 await request.json()，只是后者在内部在调用了前者拿到字节流之后、自动帮你 loads 成了字典。因此使用 await request.json() 也侧面要求，我们必须在发送请求的时候必须使用 json 参数传递（传递的是字典转成的 json、所以也能解析成字典），否则使用 await request.json() 是无法正确解析的。

路径参数、查询参数、请求体

我们可以将三者混合在一起：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptional, ListfromfastapiimportFastAPI, Request, ResponsefrompydanticimportBaseModelimportuvicornapp = FastAPI()classGirl(BaseModel):name:strage: Optional[str] =Nonelength:floathobby: List[str]@app.post("/girl/{user_id}")asyncdefread_girl(user_id,q:str,girl: Girl):return{"user_id": user_id, "q": q, **dict(girl)}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

里面我们同时指定了路径参数、查询参数和请求体，FastAPI 依然是可以正确区分的，当然我们也可以使用 Request 对象。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptional, List, DictfromfastapiimportFastAPI, Request, ResponsefrompydanticimportBaseModelimportuvicornapp = FastAPI()@app.post("/girl/{user_id}")asyncdefread_girl(user_id,request: Request):q = request.query_params.get("q")data: Dict =awaitrequest.json()data.update({"user_id": user_id, "q": q})returndataif__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

可以自己测试一下，仍然是可以正确返回的。

多个请求体参数

我们上面的只接收一个 json 请求体，如果是接收两个呢？

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptional, ListfromfastapiimportFastAPI, Request, ResponsefrompydanticimportBaseModelimportuvicornapp = FastAPI()classGirl(BaseModel):name:strage: Optional[str] =NoneclassBoy(BaseModel):name:strage:int@app.post("/boy_and_girl")asyncdefread_boy_and_girl(girl: Girl,boy: Boy):return{"girl":dict(girl), "boy":dict(boy)}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

此时在传递的时候，应该按照如下方式传递：

应该将两个 json 嵌套在一起，组成一个更大的 json，至于 key 就是我们的函数参数名。因此这种方式其实就等价于：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportOptional, List, DictfromfastapiimportFastAPI, Request, ResponsefrompydanticimportBaseModelimportuvicornapp = FastAPI()classBoyAndGirl(BaseModel):girl: Dictboy: Dict@app.post("/boy_and_girl")asyncdefread_boy_and_girl(boy_and_girl: BoyAndGirl):returndict(boy_and_girl)if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

这种方式也是可以实现的，只不过就字典内部的字典的不可进行限制了。当然啦，我们仍然可以使用 Request 对象，得到字典之后自己再进行判断，因为对于 json 而言，内部的字段可能是会变的，而且最关键的是字段可能非常多。这个时候，我个人更倾向于使用 Request 对象。

Form 表单

我们调用 requests.post，如果参数通过 data 传递的话，则相当于提交了一个 form 表单，那么在 FastAPI 中可以通过 await request.form() 进行获取，注意：内部同样是先调用 await request.body()。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, Request, Responseimportuvicornapp = FastAPI()@app.post("/girl")asyncdefgirl(request: Request):# 此时 await request.json() 报错，因为是通过 data 参数传递的，相当于 form 表单提交# 如果是通过 json 参数传递，那么 await request.form() 会得到一个空表单form =awaitrequest.form()return[form.get("name"), form.getlist("age")]if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

当然我们也可以通过其它方式：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, Formimportuvicornapp = FastAPI()@app.post("/user")asyncdefget_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() 拿到相关的表单信息。所以如果你觉得某个参数不适合类型注解，那么你可以单独通过 Request 对象进行解析。

文件上传

那么问题来了，FastAPI 如何接收用户的文件上传呢？首先如果想使用文件上传功能，那么你必须要安装一个包 python-multipart，直接pip install python-multipart 即可。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, File, UploadFileimportuvicornapp = FastAPI()@app.post("/file1")asyncdeffile1(file:bytes= File(...)):returnf"文件长度: {len(file)}"@app.post("/file2")asyncdeffile1(file: UploadFile = File(...)):returnf"文件名: {file.filename}, 文件大小: {len(awaitfile.read())}"if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

我们看到一个直接获取字节流，另一个是获取类似于文件句柄的对象。如果是多个文件上传要怎么做呢？

# -*- coding:utf-8 -*-# @Author: komeiji satorifromtypingimportListfromfastapiimportFastAPI, UploadFile, Fileimportuvicornapp = FastAPI()@app.post("/file")asyncdeffile(files: List[UploadFile] = File(...)):"""指定类型为列表即可"""foridx, finenumerate(files):files[idx] = f"文件名: {f.filename}, 文件大小: {len(awaitf.read())}"returnfilesif__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

此时我们就实现了 FastAPI 文件上传，当然文件上传并不影响我们处理表单，可以自己试一下同时处理文件和表单。

返回静态资源

下面来看看 FastAPI 如何返回静态资源，首先我们需要安装 aiofiles，直接 pip 安装即可。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIfromfastapi.staticfilesimportStaticFilesimportuvicornapp = FastAPI()# name 参数只是起一个名字，FastAPI 内部使用app.mount("/static", StaticFiles(directory=r"C:\Users\satori\Desktop\bg"), name="static")if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

浏览器输入：localhost:5555/static/1.png，那么会返回C:\Users\satori\Desktop\bg 下的 1.png 文件。

错误处理

错误处理也是一个不可忽视的点，错误有很多种，比如：

客户端没有足够的权限执行此操作客户端没有访问某个资源的权限客户端尝试访问一个不存在的资源...

这个时候我们应该将错误通知相应的客户端，这个客户端可以浏览器、代码程序、IoT 设备等等。

但是就我个人而言，更倾向于使用 Response 对象，将里面的 status_code 设置为 404，然后在返回的 json 中指定错误信息。不过 FastAPI 内部也提供了一些异常类：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, HTTPExceptionimportuvicornapp = FastAPI()@app.get("/items/{item_id}")asyncdefread_item(item_id:str):ifitem_id != "foo":# 里面还可以传入 headers 设置响应头raiseHTTPException(status_code=404, detail="item 没有发现")return{"item": "bar"}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

HTTPException 是一个普通的 Python 异常类（继承了 Exception），它携带了 API 的相关信息，既然是异常，那么我们不能 return、而是要 raise。

个人觉得这个不是很常用，至少我本人很少用这种方式返回错误，因为它能够携带的信息太少了。

自定义异常

FastAPI 内部提供了一个HTTPException，但是我们也可以自定义，但是注意：我们自定义完异常之后，还要定义一个 handler，将异常和 handler 绑定在一起，然后引发该异常的时候就会触发相应的 handler。

fromfastapiimportFastAPI, Requestfromfastapi.responsesimportORJSONResponseimportuvicornapp = FastAPI()classASCIIException(Exception):"""何もしません"""# 通过装饰器的方式，将 ASCIIException 和 ascii_exception_handler 绑定在一起@app.exception_handler(ASCIIException)asyncdefascii_exception_handler(request: Request, exc: ASCIIException):"""当引发 ASCIIException 的时候，会触发 ascii_exception_handler 的执行同时会将 request 和 exception 传过去"""returnORJSONResponse(status_code=404, content={"code":404, "message": "你必须传递 ascii 字符串"})@app.get("/items/{item_id}")asyncdefread_item(item_id:str):ifnotitem_id.isascii():raiseASCIIExceptionreturn{"item": f"get {item_id}"}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

关于 Request、Response，我们除了可以通过 fastapi 进行导入，还可以通过starlette 进行导入，因为 fastapi 的路由映射是通过starlette 来实现的。当然我们直接从 fastapi 中进行导入即可。

自定义 404

当访问一个不存在的 URL，我们应该提示用户，比如：您要找到页面去火星了。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIfromfastapi.responsesimportORJSONResponsefromfastapi.exceptionsimportStarletteHTTPExceptionimportuvicornapp = FastAPI()@app.exception_handler(StarletteHTTPException)asyncdefnot_found(request, exc):returnORJSONResponse({"code":404, "message": "您要找的页面去火星了。。。"})if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

此时当我们访问一个不存在的 URL 时，就会返回我们自定义的 JSON 字符串。

后台任务

如果一个请求耗时特别久，那么我们可以将其放在后台执行，而 FastAPI 已经帮我们做好了这一步。我们来看一下：

# -*- coding:utf-8 -*-# @Author: komeiji satoriimporttimefromfastapiimportFastAPI, BackgroundTasksimportuvicornapp = FastAPI()defsend_email(email:str, message:str= ""):"""发送邮件，假设耗时三秒"""time.sleep(3)print(f"三秒之后邮件发送给 {email!r}, 邮件信息: {message!r}")@app.get("/user/{email}")asyncdeforder(email:str, bg_tasks: BackgroundTasks):"""这里需要多定义一个参数此时任务就被添加到后台，当 Response 对象返回之后触发"""bg_tasks.add_task(send_email, email, message="这是一封邮件")# 我们在之前介绍 Response 的时候说过，里面有一个参数 background# 所以我们也可以将任务放在那里面# 因此我们还可以：# return Response(orjson.dumps({"message": "邮件发送成功"}), background=BackgroundTask(send_email, email, message="这是一封邮件"))return{"message": "邮件发送成功"}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

首先请求肯定是成功的：

然后响应的 3 秒，终端会出现如下打印：

所以此时任务是被后台执行了的，注意：任务是在响应返回之后才后台执行。

APIRouter

APIRouter 类似于 Flask 中的蓝图，可以更好的组织大型项目，举个栗子：

在我当前的工程目录中有一个 app 目录和一个 main.py，其中 app 目录中有一个 app01.py，然后我们看看它们是如何组织的。

# app/app01.pyfromfastapiimportAPIRouterrouter = APIRouter(prefix="/router")# 以后访问的时候要通过 /router/v1 来访问@router.get("/v1")asyncdefv1():return{"message": "hello world"}# main.pyfromfastapiimportFastAPIfromapp.app01importrouterimportuvicornapp = FastAPI()# 将 router 注册到 app 中，相当于 Flask 中的 register_blueprintapp.include_router(router)if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后可以在外界通过 /router/v1 的方式来访问。

中间件

中间件在 web 开发中可以说是非常常见了，说白了中间件就是一个函数或者一个类。在请求进入视图函数之前，会先经过中间件（被称为请求中间件），而在中间件里面，我们可以对请求进行一些预处理，或者实现一个拦截器等等；同理当视图函数返回响应之后，也会经过中间件（被称为响应中间件），在中间件里面，我们也可以对响应进行一些润色。

自定义中间件

在 FastAPI 里面也支持像 Flask 一样自定义中间件，但是 Flask 里面有请求中间件和响应中间件，但是在 FastAPI 里面这两者合二为一了，我们看一下用法。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, Request, Responseimportuvicornimportorjsonapp = FastAPI()@app.get("/")asyncdefview_func(request: Request):return{"name": "古明地觉"}@app.middleware("http")asyncdefmiddleware(request: Request, call_next):"""定义一个协程函数，然后使用 @app.middleware("http") 装饰，即可得到中间件"""# 请求到来时会先经过这里的中间件ifrequest.headers.get("ping", "") != "pong":response = Response(content=orjson.dumps({"error": "请求头中缺少指定字段"}),media_type="application/json",status_code=404)# 当请求头中缺少 "ping": "pong"，在中间件这一步就直接返回了，就不会再往下走了# 所以此时就相当于实现了一个拦截器returnresponse# 然后，如果条件满足，则执行 await call_next(request)，关键是这里的 call_next# 如果该中间件后面还有中间件，那么 call_next 就是下一个中间件；如果没有，那么 call_next 就是对应的视图函数# 这里显然是视图函数，因此执行之后会拿到视图函数返回的 Response 对象# 所以我们看到在 FastAPI 中，请求中间件和响应中间件合在一起了response: Response =awaitcall_next(request)# 这里我们在设置一个响应头response.headers["status"] = "success"returnresponseif__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

我们可以测试一下：

测试结果也印证了我们的结论。

内置的中间件

通过自定义中间件，我们可以在不修改视图函数的情况下，实现功能的扩展。但是除了自定义中间件之外，FastAPI 还提供了很多内置的中间件。

app = FastAPI()# 要求请求协议必须是 https 或者 wss，如果不是，则自动跳转fromstarlette.middleware.httpsredirectimportHTTPSRedirectMiddlewareapp.add_middleware(HTTPSRedirectMiddleware)# 请求中必须包含 Host 字段，为防止 HTTP 主机报头攻击，并且添加中间件的时候，还可以指定一个 allowed_hosts，那么它是干什么的呢？# 假设我们有服务 , , # 但我们不希望用户访问 ，就可以像下面这么设置，如果指定为 ["*"]，或者不指定 allow_hosts，则表示无限制fromstarlette.middleware.trustedhostimportTrustedHostMiddlewareapp.add_middleware(TrustedHostMiddleware, allowed_hosts=["", ""])# 如果用户的请求头的 Accept-Encoding 字段包含 gzip，那么 FastAPI 会使用 GZip 算法压缩# minimum_size=1000 表示当大小不超过 1000 字节的时候就不压缩了fromstarlette.middleware.gzipimportGZipMiddlewareapp.add_middleware(TrustedHostMiddleware, minimum_size=1000)

除了这些，还有其它的一些内置的中间件，可以自己查看一下，不过不是很常用。

CORS

CORS 过于重要，我们需要单独拿出来说。

CORS（跨域资源共享）是指浏览器中运行的前端里面拥有和后端通信的 JavaScript 代码，而前端和后端处于不同源的情况。源：协议（http、https）、域（、、localhost）以及端口（80、443、8000），只要有一个不同，那么就是不同源。比如下面都是不同的源：

http://localhosthttps://localhosthttp://localhost:8080

即使它们都是 localhost，但是它们使用了不同的协议或端口，所以它们是不同的源。假设你的前端运行在 localhost:8080，并且尝试与 localhost:5555 进行通信；然后浏览器会向后端发送一个 HTTP OPTIONS 请求，后端会发送适当的 headers 来对这个源进行授权；所以后端必须有一个 "允许的源" 列表，如果前端对应的源是被允许的，浏览器才会允许前端向后端发请求，否则就会出现跨域失败。

而默认情况下，前后端必须是在同一个源，如果不同源那么前端就会请求失败。而前后端分离早已成为了主流，因此跨域问题是必须要解决的。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewareimportuvicornapp = FastAPI()app.add_middleware(CORSMiddleware,# 允许跨域的源列表，例如 [""] 等等，["*"] 表示允许任何源allow_origins=["*"],# 跨域请求是否支持 cookie，默认是 False，如果为 True，allow_origins 必须为具体的源，不可以是 ["*"]allow_credentials=False,# 允许跨域请求的 HTTP 方法列表，默认是 ["GET"]allow_methods=["*"],# 允许跨域请求的 HTTP 请求头列表，默认是 []，可以使用 ["*"] 表示允许所有的请求头# 当然 Accept、Accept-Language、Content-Language 以及 Content-Type 总之被允许的allow_headers=["*"],# 可以被浏览器访问的响应头, 默认是 []，一般很少指定# expose_headers=["*"]# 设定浏览器缓存 CORS 响应的最长时间，单位是秒。默认为 600，一般也很少指定# max_age=1000)if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

以上即可解决跨域问题。

高阶操作

下面我们看一些 FastAPI 的高阶操作，这些操作有的不一定能用上，但用上了确实会方便许多。

其它的响应

返回 json 数据可以是：JSONResponse、UJSONResponse、ORJSONResponse，Content-Type 是 application/json；返回 html 是HTMLResponse，Content-Type 是 text/html；返回PlainTextResponse，Content-Type 是 text/plain。但是我们还可以有三种响应，分别是返回重定向、字节流、文件。

重定向

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIfromfastapi.responsesimportRedirectResponseimportuvicornapp = FastAPI()@app.get("/index")asyncdefindex():returnRedirectResponse("")if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

页面中访问 /index 会跳转到 bilibili。

字节流

返回字节流需要使用异步生成器的方式：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponseimportuvicornapp = FastAPI()asyncdefsome_video():foriinrange(5):yieldf"video {i} bytes ".encode("utf-8")@app.get("/index")asyncdefindex():returnStreamingResponse(some_video())if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

如果有文件对象，那么也是可以直接返回的。

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponseimportuvicornapp = FastAPI()@app.get("/index")asyncdefindex():returnStreamingResponse(open("main.py", encoding="utf-8"))if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

文件

返回文件的话，还可以通过 FileResponse：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIfromfastapi.responsesimportFileResponseimportuvicornapp = FastAPI()@app.get("/index")asyncdefindex():# filename 如果给出，它将包含在响应的 Content-Disposition 中。returnFileResponse("main.py", filename="这不是main.py")if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

可以自己发请求测试一下。

HTTP 验证

如果当用户访问某个请求的时候，我们希望其输入用户名和密码来确认身份的话该怎么做呢？

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPI, Dependsfromfastapi.securityimportHTTPBasic, HTTPBasicCredentialsimportuvicornapp = FastAPI()security = HTTPBasic()@app.get("/index")asyncdefindex(credentials: HTTPBasicCredentials = Depends(security)):return{"username": credentials.username, "password": credentials.password}if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

当用户访问 /index 的时候，会提示输入用户名和密码：

输入完毕之后，信息会保存在credentials，我们可以获取出来进行验证。

websocket

然后我们来看看 FastAPI 如何实现 websocket：

# -*- coding:utf-8 -*-# @Author: komeiji satorifromfastapiimportFastAPIfromfastapi.websocketsimportWebSocketimportuvicornapp = FastAPI()@app.websocket("/ws")asyncdefws(websocket: WebSocket):awaitwebsocket.accept()whileTrue:# websocket.receive_bytes()# websocket.receive_json()data =awaitwebsocket.receive_text()awaitwebsocket.send_text(f"收到来自客户端的回复: {data}")if__name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后我们通过浏览器进行通信：

<!DOCTYPEhtml><html lang="en"><head><meta charset="UTF-8"><title>Title</title></head><body><script>ws =newWebSocket("ws://localhost:5555/ws");//如果连接成功, 会打印下面这句话, 否则不会打印ws.onopen =function() {console.log('连接成功')};//接收数据, 服务端有数据过来, 会执行ws.onmessage =function(event) {console.log(event)};//服务端主动断开连接, 会执行.//客户端主动断开的话, 不执行ws.onclose =function() { }</script></body></html>

FastAPI 服务的部署

目前的话，算是介绍了 FastAPI 的绝大部分内容，然后我们来看看 FastAPI 服务的部署，其实部署很简单，直接 uvicorn.run 即可。但是这里面有很多的参数，我们主要是想要介绍这些参数。

defrun(app, **kwargs):config = Config(app, **kwargs)server = Server(config=config)......

我们看到 app 和 **kwargs 都传递给了 Config，所以我们只需要看 Config 里面都有哪些参数即可。这里选出一部分：

app：第一个参数，不需要解释host：监听的ipport：监听的端口uds：绑定的 unix domain socket，一般不用fd：从指定的文件描述符中绑定 socketloop：事件循环实现，可选项为 auto|asyncio|uvloop|iocphttp：HTTP 协议实现，可选项为 auto|h11|httptoolsws：websocket 协议实现，可选项为 auto|none|websockets|wsprotolifespan：lifespan 实现，可选项为 auto|on|offenv_file：环境变量配置文件log_config：日志配置文件log_level：日志等级access_log：是否记录日志use_colors：是否带颜色输出日志信息interface：应用接口，可选 auto|asgi3|asgi2|wsgidebug：是否开启 debug 模式reload：是否自动重启reload_dirs：要自动重启的目录reload_delay：多少秒后自动重启workers：工作进程数limit_concurrency：并发的最大数量limit_max_requests：能 hold 住的最大请求数

小结

总的来说，FastAPI 算是当前最流行的异步框架了，并且它完全可以在生产中使用，是值得信赖的。当然使用异步框架，最重要的是要搭配一个异步驱动去访问数据库，因为 web 服务的瓶颈都是在数据库上面。