FastAPI - 一款新型的 Python Web 框架(对比 Flask)

Posted 2023-02-15

参考技术A 近日只是为了想尽办法为 Flask 实现 Swagger UI 文档功能，基本上要让 Flask 配合 Flasgger, 所以写了篇 Flask 应用集成 Swagger UI 。然而不断的 Google 过程中偶然间发现了 FastAPI 这么一款集成了 Swagger UI 的更新的 Python Web 框架。起初想要在标题中表达的意思大概是 Flask + Swagger = FastAPI, 后来发现 FastAPI 的闪亮点不仅如此，顺便找了些 Flask 与 FastAPI 对比的文章来，在文后附有链接。

本文不对 Flask 与 FastAPI 的各个方面对进行对比，本人兴趣依然还在 FastAPI 的 Swagger UI 功能，以及与 Flask 的 Blueprint 类似的特性。如果要拿 Flask 与 FastAPI 比较的话，应该用 Flask 2.x, 因为它开始支持类似 @app.get 的装饰器，并引入了 async 路由函数。

Flask 是在 2010 年发布的，它构建于 WSGI(Python Web Server Gateway Interface) 之上的，产品环境中运行需与 uWSGI, Gunicorn 搭配，或用 mod_wsgi 模块与 Apache 集成。因发布较早，所以目前应该有较多的使用者。Flask 2.0 需要 Python 3.6+ 的支持，如果支持 async , 需 Python 3.7+

FastAPI 发布于 2018 年，构建于 ASGI(Asynchronous Server Gateway Interface) 之上，在 IO 密集型的应用中有更优越的性能。生成环境中配合 ASGI 服务器，如 Uvicorn 或 Hypercorn . FastAPI 最为亮丽的特性是集成了 Swagger UI -- 外加一个福利 ReDoc 。FastAPI 需 Python 3.6+ 版本。

毕竟是在开始学一个新的框架，还是从它的基本用法开始，途中会穿插与 Flask 的对比。

安装:

当前安装的 fastapi 版本为 0.70.1, uvicorn 版本为 0.16.0。开始第一个例子，摘自官方

创建一个 main.py 文件，内容为

注：以上两个函数前面可以加上 async 关键字

启动服务，用命令

注： uvicorn --help 列出详细帮助，如启动多少个 worker, 绑定网络接口和端口号, 配置 SSL 证书等。

访问服务

另一个服务就是 http://localhost:8000/items/100?q=somequery

与 Flask 的对比

回顾一个 Flask 启动服务的方式有

如果程序文件名为 app.py 或 wsgi.py 的话，可以省略 FLASK_APP 环境变量，直接执行 flask run 就行。另一种方式就是在代码中加上 main 入口

然后像普通 Python 代码一样执行

对于 FastAPI, 如果总是要用 uvicorn 来启动服务的话，在 IDE 中调用就变得不那么便利。由于 uvicorn 本身就是 Python 实现，当然也就可以把命令过程写到 Python 代码中

注：uvicorn.run() 可接受更多的参数，相当于 uvicorn --help 可接受的。

同样把 main.py 当作普通 Python 程序来启动即可

前面讲过，FastAPI 进入我的视野最首要的一个原因就是它集成了 Swagger UI, 在启动了 FastAPI 服务后，只要访问 http://localhost:8000/docs, 熟悉的 Swagger UI 即刻映入眼帘

同时注意到函数映射为 API 的名称，由于参数是带有类型提示的，所以在 SwaggerUI 中也能看到相应的类型。其他定制 Swagger UI 的工作需要时可细细研究。

FastAPI 除了 SwaggerUI 外，还同时给了我们另一个选择，那就是 redoc, 访问 http://localhost:8000/redoc, 看到下面的界面

我也是头一回从 FastAPI 这里了解到还有一个 Redoc 的 Web API 文档工具，它和 SwaggerUI 类似，都是支持 Open API。

使用 FastAPI 时，在 API 文档中产生请求/响应的 Model 类型也很简单，下面的例子同时包含自定义的输入输出类型

下面是 main.py 的全新内容

再查看 http://localhost:8000/docs 文档，看到 PUT /items/item_id 的界面如下

我们能看到请求与响应中的 Example Value 和 Schema 定义

梳理完 FastAPI 的自动 API 文档功能，进入下一个话题

FastAPI 类似于 Flask 的 Blueprints 功能没有一个特定的名称，只笼统的给出解决方案 Bigger Applications - Multiple Files . 我们来看一下它是怎么实现的。这里简化官方的例子，排除了 dependencies 的应用，要用到的目录结构为

每个目录中都可以放一个 __init__.py 文件，标识为包，接着是每个文件的内容

app/internal/admin.py

app/routers/items.py

app/routers/users.py

在每个单独定义路由的文件里用到的是 APIRouter , 它相当是一个迷你的 FastAPI , 也类似于 Flask 的 Blueprint . 上面分别演示了两种 APIRouter 的声明方式，声明时带不带 prefix 和 tags。同时在为函数定义路由时也可指定 tags，在后面我们将会注意到 tags 只是为 API 文档分类用的。

app/main.py

这里实际上使用了三种定义路径前缀与 tags(API 分类) 的方式

现在运行

查看 SwaggerUI http://localhost:8000/docs

到目前为止我重点关注 FastAPI 的就是以上那两个特性：自动 API 文档与大程序多文件(Blueprint) 功能。其他的特性可查阅 FastAPI 的 Features 。比如：

原文 https://yanbin.blog/fastapi-new-python-web-framework-vs-flask/

三分钟了解 Python3 的异步 Web 框架 FastAPI

快速编码，功能完善。从启动到部署，实例详解异步 py3 框架选择 FastAPI 的原因。

FastAPI 介绍

FastAPI 与其它 Python-Web 框架的区别

在 FastAPI 之前，Python 的 Web 框架使用的是 django、flask、tornado 三种 Web 框架。

• django 自带 admin，可快速构建，但是比较笨重。如果是 mvc 形式的开发，很多已经封装好了，的确蛮合适。但如果是 restful 风格设计，则 django 就显得有一些笨重了。

• flask 快速构建，自由度高。因为它十分轻盈，插件即插即用，很适合用来做 restful 风格的设计

• tornado Python Web 框架和异步网络库，它执行非阻塞 I/O , 没有对 REST API 的内置支持，但是用户可以手动实现。

• FastAPI 快速构建，异步 IO，自带 Swagger 作为 API 文档，不用后续去内嵌 Swagger-Ui

我个人认为 FastAPI 是一个专门为 restful 风格设计，全面服务于 API 形式的 Web 后端框架。

FastAPI 官方定位

在 FastAPI 官方文档中，可以看到官方对 FastAPI 的定位：

• 快速：非常高的性能，向 NodeJS 和 go 看齐（感谢 Starlette 和 Pydantic)

• 快速编码：将功能开发速度提高约 200％ 至 300％。

• 错误更少：减少约 40％ 的人为错误（开发人员）。* (FastAPI 内置很多 Python 高版本的语法，比如类型注释，typing 库等等，因此被要求的 Python 版本为 3.6+)

• 简易：旨在易于使用和学习。减少阅读文档的时间。

• 功能完善: 自带 Swagger 作为 API 文档

Framework Benchmarks

https://www.techempower.com/benchmarks/#section=data-r19&hw=ph&test=fortune

上图可以看出，在高并发下 4 个框架的排名情况。单纯从性能出发，Web 框架是排在第一的。在选用框架的时候，性能是一方面，我们还要看业务上的需求和使用场景，最适合的才是最好的。

下面简单介绍一下 FastAPI 的一些用法和特性.

启动FastAPI

1 # pip install fastapi
2 # pip install uvicorn
3 from fastapi import FastAPI
4 app = FastAPI()
5 @app.get("/")
6 def read_root():
7 return {"Hello": "World"}
8 @app.get("/items/{item_id}")
0 def read_item(item_id: int, q: str = None):
10 return {"item_id": item_id, "q": q}
11 # uvicorn main:app # 启动
12 # uvicorn main:app --reload # 支持热更新
13 # uvicorn main:app --host 0.0.0.0 --port 8889 --reload # 自定义IP+端口
14

FastAPI 支持异步请求

1 from fastapi import FastAPI
2 app = FastAPI()
3 @app.get("/")
4 async def read_root():
5 return {"Hello": "World"}
6
7 @app.get("/items/{item_id}")
8 async def read_item(item_id: int, q: str = None):
9 return {"item_id": item_id, "q": q}
10

对 API 接口的支持性优异

设置根目录

1 # main.py
2 from fastapi import FastAPI
3 import users
4 app = FastAPI()
5 app.include_router(
6 users.router,
7 prefix="/fastapi/play/v1/users", # 路由前缀
8 tags=[‘users‘] # 路由接口类别
9 )
10 # routers/users.py
11 from fastapi import FastAPI,APIRouter
12 from datetime import datetime,timedelta
13 router = APIRouter()
14 @router.get("/get/users/")
15 async def get_users():
16 return {
17 "desc":"Return to user list"
18 }
19

对路径参数进行限制

1 # 根据名字获取列表
2 @router.get("/get/user/{username}")
3 async def get_user_by_username(username :str):
4 """
5 - username: 用户名
6 """
7 return {
8 "desc":"this username is "+ username
9 }
10

对查询参数做限制

1 @router.get("/friends/")
2
3 # 设置为None的时候，默认不可以不填
4 async def get_friends_by_id(id :int=None):
5 for item in test_json[‘friends‘]:
6 if item[‘id‘] == id:
7 return item
8 else:
9 return {
10 "desc": "no this id"
11 }
12 # 多参数请求查询
13 from typing import List
14 @router.get("/items/")
15 async def read_items(q: List[str] = Query(["foo", "bar"])):
16 query_items = {"q": q}
17 return query_items
18

设置请求体

1 # 设置请求实体
2 from pydantic import BaseModel,Field
3 class requestModel(BaseModel):
4 name :str
5 age : int = Field(..., gt=0, description="The age must be greater than zero")
6 desc: str
7
8
9 @router.post("/post/UserInfo/")
10 async def post_UserInfo(item: requestModel):
11 return item
12

请求体嵌套

1 from pydantic import BaseModel,Field
2 class levelInfoModel(BaseModel):
3 id:int = None
4 info: str = None
5
6 class ClassInfo(BaseModel):
7 id: int = None
8 name: str = Field(..., max_length=20, min_length=10,
9 description="The necessary fields")
10 desc: str = Field(None, max_length=30, min_length=10)
11 levelInfo: List[levelInfoModel]
12
13 class Config:
14 schema_extra = {
15 "example": {
16 "id": 1,
17 "name": "Foo",
18 "desc": "A very nice Item",
19 "levelInfo": [{
20 "id": 1,
21 "info": "一级"
22 }]
23 }
24 }
25
26 @router.post("/info/")
27 async def get_classInfo(item:ClassInfo):
28 return item
29

自定义响应码

1 @router.post("/items/", status_code=201)
2 async def create_item(name: str):
3 return {"name": name}
4
5 from fastapi import FastAPI, status
6
7
8 @app.post("/items/", status_code=status.HTTP_201_CREATED)
9 async def create_item(name: str):
10 return {"name": name}
11

依赖注入

1 from fastapi import Depends, FastAPI
2
3 async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
4 return {"q": q, "skip": skip, "limit": limit}
5
6 @router.get("/items/")
7 async def read_items(commons: dict = Depends(common_parameters)):
8 return commons
9
10 @router.get("/users/")
11 async def read_users(commons: dict = Depends(common_parameters)):
12 return commons
13

FastAPI 框架支持多层嵌套依赖注入

登录demo

1 # 安装环境
2 mkdir fastapi-demo && cd fastapi-demo
3 virtualenv env
4 source env/bin/activate
5
6 # 下载项目
7 git clone https://github.com/hzjsea/BaseFastapi
8 cd BaseFastapi/
9 pip install -r requirements.txt
10 # 开启项目
11 uvicorn main:app --reload
12 # uvicorn main:app --host 0.0.0.0 --port 80 --reload
13

总结

FastAPI 的设计还是很符合 restful 的，在用到很多新技术的同时，也没有抛弃之前一些比较好用的内容，包括类型注释、依赖注入，Websocket，swaggerui 等等，以及其它的一些注释，比如 GraphQL。

数据库以及 orm 的选择

• sqlalchemy 但是不支持异步，不过貌似可以扩展成异步。

• tortoise-orm 类 django-orm 的异步 orm，不过正在起步过程中，有些功能还没有完成。

sqlalchemy实例

1 from typing import List
2 import databases
3 import sqlalchemy
4 from fastapi import FastAPI
5 from pydantic import BaseModel
6 # SQLAlchemy specific code, as with any other app
7 DATABASE_URL = "sqlite:///./test.db"
8 # DATABASE_URL = "postgresql://user:password@postgresserver/db"
9 database = databases.Database(DATABASE_URL)
10 metadata = sqlalchemy.MetaData()
11 notes = sqlalchemy.Table(
12 "notes",
13 metadata,
14 sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
15 sqlalchemy.Column("text", sqlalchemy.String),
16 sqlalchemy.Column("completed", sqlalchemy.Boolean),
17 )
18 engine = sqlalchemy.create_engine(
19 DATABASE_URL, connect_args={"check_same_thread": False}
20 )
21 metadata.create_all(engine)
22
23
24 class NoteIn(BaseModel):
25 text: str
26 completed: bool
27
28
29 class Note(BaseModel):
30 id: int
31 text: str
32 completed: bool
33
34
35 app = FastAPI()
36
37
38 @app.on_event("startup")
39 async def startup():
40 await database.connect()
41
42
43 @app.on_event("shutdown")
44 async def shutdown():
45 await database.disconnect()
46
47
48 @app.get("/notes/", response_model=List[Note])
49 async def read_notes():
50 query = notes.select()
51 return await database.fetch_all(query)
52
53
54 @app.post("/notes/", response_model=Note)
55 async def create_note(note: NoteIn):
56 query = notes.insert().values(text=note.text, completed=note.completed)
57 last_record_id = await database.execute(query)
58 return {**note.dict(), "id": last_record_id}
59

tortoise-orm实例

1 # main.py
2 from tortoise.contrib.fastapi import HTTPNotFoundError, register_tortois
3 # 创建的数据表
4 models = [
5 "app.Users.models",
6 "app.Face.models",
7 "app.Roster.models",
8 "app.Statistical.models",
9 "app.pay.models"
10 ]
11
12 register_tortoise(
13 app,
14 db_url="mysql://username:password@ip:port/yydb",
15 modules={"models": models},
16 generate_schemas=True,
17 add_exception_handlers=True,
18 )
19
20 # models.py
21 from tortoise import fields,models
22 from tortoise.contrib.pydantic import pydantic_queryset_creator
23 from pydantic import BaseModel
24 class RosterGroupTable(models.Model):
25 id = fields.IntField(pk=True)
26 phone = fields.CharField(max_length=20,blank=True,null=True)
27 name = fields.CharField(max_length=20)
28
29 class Meta:
30 db = "RosterGroupTable"
31
32 class RosterTabel(models.Model):
33 id = fields.IntField(pk=True)
34 phone = fields.CharField(max_length=20,blank=True,null=True)
35 name = fields.CharField(max_length=20)
36 group_id = fields.ForeignKeyField(model_name=‘models.RosterGroupTable‘,on_delete=fields.CASCADE,related_name="events",blank=True,null=True)
37
38 class Meta:
39 db = "RosterTabel"
40
41 RosterGroupTable_desc = pydantic_queryset_creator(RosterGroupTable)
42 RosterTabel_desc = pydantic_queryset_creator(RosterTabel)
43
44
45
46 # roster.py
47 @router.post("/roster_statistics/add")
48 async def roster_add_statics(*,item:RosterItem,token :str):
49 res = await RosterTabel.filter(id=item[‘memberId‘]).first()
50 if res:
51 await StatisticalRosterTable.create(
52 phone = current_user.Phone,
53 date = item[‘date‘],
54 time = item[‘time‘],
55 data = item[‘data‘],
56 name_id_id = item[‘memberId‘],
57 temp_type = item[‘tm_type‘],
58 today_num = item[‘todayNum‘],
59 group_id_id = res.group_id_id,
60 name = res.name
61 )
62 else:
63 return rp_faildMessage(msg="名单不存在",code=-1)
64 return rp_successMessage(msg="名单创建成功",code=0)
65

部署

dockerfile

1 #================================================================================
2 # 基于Python3.7的创建fastapi的dockerfile文件
3 # fastapi + Python3.7 + guvicorn
4 #================================================================================
5
6 FROM Python:3.7
7 LABEL version="v1"
8 description="fastapi"
9 maintainer="hzjsea"
10 using="fastapi & Python3.7 office image & gunicorn"
11 WORKDIR /root/code
12 COPY . .
13 RUN pip install -r requirements.txt
14 EXPOSE 8889
15 CMD ["gunicorn","-c","guvicorn.conf","main:app"]
16

supervisor项目托管

1 [program:webserver]
2 directory=/root/hzj/fastapi_play
3 command=/root/hzj/pro_env_all/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8888
4 autostart = true
5

部署完整示例

FastAPI官方提供了一个前后端分离项目完整示例
https://github.com/tiangolo/full-stack-fastapi-postgresql

文档及项目地址：

Documentation: https://fastapi.tiangolo.com

推荐阅读

从 301 跳转，聊聊边缘规则的那些小妙用

从新冠疫情出发，漫谈 Gossip 协议