首頁>技術>

（進階篇）簡析一個比Flask和Tornado更高效能的API 框架FastAPI

前言

上一篇已經初步了解了 FastAPI 的基本使用，但是如果想要真正把 FastAPI 部署上線到伺服器，那麼你需要了解更多，學習更多。所以本篇內容將注重於 FastAPI 的專案生產環境，諸如 資料庫，路由藍圖，資料驗證等問題在 FastAPI 中的具體操作和一些自己碰到的坑，分享給正在進攻 FastAPI 的各位小夥伴。

藍圖

事實上，FastAPI 並沒有關於藍圖 (Blueprint) 的定義，在 FastAPI 中使用 Include_route 方法來新增路由，也就是我們所熟知的藍圖了。

import timefrom typing import Listfrom starlette.templating import Jinja2Templatesfrom fastapi import Depends, FastAPI, HTTPExceptionfrom starlette.staticfiles import StaticFilesfrom starlette.templating import Jinja2Templatesfrom app import modelsfrom app.database.database import SessionLocal, enginefrom app.home import user,indexapp = FastAPI()app.mount("/static", StaticFiles(directory="app/static"), name="static") # 掛載靜態檔案，指定目錄templates = Jinja2Templates(directory="templates") # 模板目錄app.include_router(index.userRouter)app.include_router(user.userRouter,prefix="/user")

可以看到在 home 目錄引入了 user.py 和 index.py 檔案，注意必須要在檔案中初始化一個 APIRouter() 類物件 (當然如果需要，可以選擇繼承)，prefix 指明子路由的路徑，更多的引數使用請參考官方文件。

# user.pyfrom starlette.templating import Jinja2Templatesfrom app import schemas, modelsfrom app.database.database import get_dbfrom app.home import crudfrom fastapi import Depends, HTTPException, Formfrom sqlalchemy.orm import Sessionfrom app.models import Userfrom sqlalchemy.orm import Sessionfrom fastapi import APIRouter, HTTPException,Requestfrom fastapi.responses import RedirectResponseuserRouter = APIRouter()templates = Jinja2Templates(directory="app/templates") # 模板目錄@userRouter.post("/login/", response_model=schemas.UserOut)async def login(*,request: Request,db: Session = Depends(get_db), username: str = Form(None), password: str = Form(None),):    if request.method == "POST":        db_user = db.query(models.User).filter(User.username == username).first()        if not db_user:            raise HTTPException(status_code=400, detail="使用者不存在")        print("驗證通過 ！！！")        return RedirectResponse('/index')    return templates.TemplateResponse("user/login.html", {"request": request})

看起來比 Flask 新增藍圖要輕鬆許多。

同時支援多種請求方式

在上面的 login 例子可以發現，我在上下文 request 中通過判斷路由的請求方式來進行響應的邏輯處理，比如如果不是 Post請求 就把它重定向到 login 頁面等等。那麼就需要同時支援多種請求方式了，巧合的是，我在 FastAPI 文件中找不到相應的說明，剛開始的時候我也迷糊了一陣。所以，只能幹原始碼了。

直接進入 APIRouter 類所在的檔案，發現新大陸。

在 APIRouter 下有個叫 add_api_route 的方法，支援 http方法 以列表的形式作為引數傳入，所以就換成了下面這種寫法：

async def login(*,request: Request,db: Session = Depends(get_db), username: str = Form(None), password: str = Form(None),):    if request.method == "POST":        db_user = db.query(models.User).filter(User.username == username).first()        if not db_user:            raise HTTPException(status_code=400, detail="使用者不存在")        print("驗證通過 ！！！")        return RedirectResponse('/index')    return templates.TemplateResponse("user/login.html", {"request": request})async def userList(*,request: Request,db: Session = Depends(get_db)):    userList = db.query(models.User).all()    return templates.TemplateResponse("user/user-index.html", {"request": request,'userList':userList})userRouter.add_api_route(methods=['GET','POST'],path="/login",endpoint=login)userRouter.add_api_route(methods=['GET','POST'],path="/list",endpoint=userList)

其中，methods 是非常熟悉的字眼，寫入你想要的 http請求方式，path 指訪問時的路徑，endpoint 就是後端方法了。

這樣就解決了同時存在於多個 http請求方式 的問題啦，編碼也更為直觀簡潔。

資料庫

在 FastAPI 中，我們一如既往的使用了 SQLAlchemy

初始化資料庫檔案：

from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker# 建立資料庫連線URISQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@127.0.0.1:3306/blog"# 初始化engine = create_engine(    SQLALCHEMY_DATABASE_URL)# 建立DBSession型別SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 建立基類 用於繼承  也可以放到初始化檔案中Base = declarative_base()# 獲取資料庫會話，用於資料庫的各種操作def get_db():    db = SessionLocal()

資料庫模型檔案：

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, DateTime, Textfrom sqlalchemy.orm import relationshipfrom datetime import datetimefrom flask_login import  UserMixinimport uuidfrom app.database.database import Baseclass User(UserMixin,Base):    __tablename__ = 'user'    id = Column(Integer, primary_key=True)    email = Column(String(64),)    username = Column(String(64), )    role = Column(String(64), )    password_hash = Column(String(128))    head_img = Column(String(128), )    create_time  = Column(DateTime,default=datetime.now)    def __repr__(self):        return '<User %r>' % self.username# 文章表class Article(Base):    __tablename__ = 'article'    id = Column(Integer, primary_key=True)    title=Column(String(32))    author =Column(String(32))    img_url = Column(Text,nullable=False)    content=Column(Text,nullable=False)    tag=Column(String(64),nullable=True)    uuid = Column(Text,default=uuid.uuid4())    desc = Column(String(100), nullable=False)    create_time = Column(DateTime,default=datetime.now)    articleDetail = relationship('Article_Detail', backref='article')    def __repr__(self):        return '<Article %r>' % self.title

增

async def articleDetailAdd(*,request: Request,db: Session = Depends(get_db),d_content:str,uid:int):    if request.method == "POST":        addArticleDetail= Article_Detail(d_content=d_content,uid=uid)        db.add(addArticleDetail)        db.commit()        db.refresh(addArticleDetail)        print("新增成功 ！！！")        return "新增成功"    return "缺少引數"

刪

async def articleDetailDel(*,request: Request,db: Session = Depends(get_db),aid:int):    if request.method == "POST":        db.query(Article_Detail).filter(Article_Detail.id == aid).delete()        db.commit()        print("刪除成功 ！！！")        return "刪除成功"    return "缺少引數"

改

async def articleDetailUpdate(*,request: Request,db: Session = Depends(get_db),aid:int,d_content:str):    if request.method == "POST":        articleInfo= db.query(Article_Detail).filter(Article_Detail.id == aid).first()        print(articleInfo)        if not articleInfo:            raise HTTPException(status_code=400, detail="no no no !!")        articleInfo.d_content = d_content        db.commit()        print("提交成功 ！！！")        return "更新成功"    return "缺少引數"

查

async def articleDetailIndex(*,request: Request,db: Session = Depends(get_db),):    articleDetailList = db.query(models.Article_Detail).all()    return templates.TemplateResponse("articleDetail/articleDetail-index.html", {"request": request,"articleDetailList":articleDetailList})

這裡是一些示例的 crud，真正部署的時候可不能這麼魯莽哇，錯誤的捕捉，資料庫的回滾，語句必須嚴謹。

資料驗證

在路由方法中，有個叫 response_model 的引數，用於限制路由方法的返回欄位。

官方文件例項：

from fastapi import FastAPIfrom pydantic import BaseModel, EmailStrapp = FastAPI()class UserIn(BaseModel):    username: str    password: str    email: EmailStr    full_name: str = Noneclass UserOut(BaseModel):    username: str    email: EmailStr    full_name: str = None@app.post("/user/", response_model=UserOut)async def create_user(*, user: UserIn):    return user

意思是 UserIn 作為請求體引數傳入，返回時必須滿足 UserOut 模型。

場景的話，可以想象使用者登陸時需要傳入使用者名稱和密碼，使用者登陸成功之後在首頁上展示使用者名稱的郵件，不展示密碼。嗯，這樣就合理了。

所以在資料庫操作的時候，可以自己定義傳入和返回的模型欄位來做有效的限制，你只需要繼承 pydantic 中的 BaseModel 基類即可，看起來是那麼的簡單合理。

異常處理

在各種 http資源 不存在或者訪問異常的時候都需要有 http狀態碼 和 異常說明，例如， 404 Not Found 錯誤，Post請求出現的 422，服務端的 500 錯誤，所以如何在程式中合理的引發異常，就變得格外重要了。

看看 FastAPI 中如何使用異常處理

from fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@app.get("/items/{item_id}")async def read_item(item_id: str):    if item_id not in items:        raise HTTPException(status_code=404, detail="Item not found")    return {"item": items[item_id]}

使用 HTTPException，傳入狀態碼 和 詳細說明，在出現邏輯錯誤時丟擲異常。

改寫 HTTPException

from fastapi import FastAPI, Requestfrom fastapi.responses import JSONResponseclass UnicornException(Exception):    def __init__(self, name: str):        self.name = nameapp = FastAPI()@app.exception_handler(UnicornException)async def unicorn_exception_handler(request: Request, exc: UnicornException):    return JSONResponse(        status_code=418,        content={"message": f"我家熱得快炸了..."},    )@app.get("/unicorns/{name}")async def read_unicorn(name: str):    if name == "yolo":        raise UnicornException(name=name)    return {"unicorn_name": name}

UnicornException 繼承自 Python 自帶的 Exception 類，在出現服務端錯誤時丟擲 418 錯誤，並附上錯誤說明。

自定義自己的異常處理程式碼

from fastapi import FastAPI, HTTPExceptionfrom fastapi.exceptions import RequestValidationErrorfrom fastapi.responses import PlainTextResponsefrom starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)async def http_exception_handler(request, exc):    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)@app.exception_handler(RequestValidationError)async def validation_exception_handler(request, exc):    return PlainTextResponse(str(exc), status_code=400)@app.get("/items/{item_id}")async def read_item(item_id: int):    if item_id == 3:        raise HTTPException(status_code=418, detail="開空調啊")    return {"item_id": item_id}

合理的使用異常處理機制，能讓專案程式碼更健壯，客戶端更友好，也易於維護。

還有嗎？

在茫茫的 FastAPI 文件中我儘可能摸索出一些易用，實用，好用的功能來和大家分享，並嘗試投入到實際的生產環境中，在這個過程中去學習更多的東西，體驗更好的服務效能。

FastAPI 官方文件十分的龐大，有非常多的地方還沒有普及和深入，比如 FastAPI 的安全加密，中介軟體的使用，應用部署等等。哈，來日方長 !!!

需要學習更多關於FastAPI 知識的話，可以戳閱讀全部，獲取詳情：

參考文件：https://fastapi.tiangolo.com/tutorial

209

Flask

資料庫

最新評論

相關內容

  • ∧ BSA-TRITC(10mg/ml) TRITC-BSA 牛血清白蛋白改性標記羅丹明
  • ∨ （完結篇）簡析一個比Flask和Tornado更高效能的API 框架FastAPI