三分钟了解 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

www.techempower.com/benchmarks/…

上图能够看出，在高并发下 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 github.com/hzjsea/Base… 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官方提供了一个先后端分离项目完整示例 github.com/tiangolo/fu…

文档及项目地址：

Documentation: fastapi.tiangolo.com

推荐阅读

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

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