三分鐘瞭解 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:[email protected]/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:[email protected]: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
