三分鐘瞭解 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 協議