|
前言 上一篇已經(jīng)初步了解了 FastAPI 的基本使用,，但是如果想要真正把 FastAPI 部署上線到服務(wù)器,，那么你需要了解更多，學(xué)習(xí)更多,。所以本篇內(nèi)容將注重于 FastAPI 的項目生產(chǎn)環(huán)境,，諸如 數(shù)據(jù)庫，路由藍圖,，數(shù)據(jù)驗證等問題在 FastAPI 中的具體操作和一些自己碰到的坑,，分享給正在進攻 FastAPI 的各位小伙伴,。 
事實上，F(xiàn)astAPI 并沒有關(guān)于藍圖 (Blueprint) 的定義,，在 FastAPI 中使用 Include_route 方法來添加路由,，也就是我們所熟知的藍圖了。 import time
from typing import List
from starlette.templating import Jinja2Templates
from fastapi import Depends, FastAPI, HTTPException
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
from app import models
from app.database.database import SessionLocal, engine
from app.home import user,index
app = FastAPI()
app.mount("/static", StaticFiles(directory="app/static"), name="static") # 掛載靜態(tài)文件,，指定目錄
templates = Jinja2Templates(directory="templates") # 模板目錄
app.include_router(index.userRouter)
app.include_router(user.userRouter,prefix="/user")
可以看到在 home 目錄引入了 user.py 和 index.py 文件,，注意必須要在文件中初始化一個 APIRouter() 類對象 (當(dāng)然如果需要，可以選擇繼承),，prefix 指明子路由的路徑,，更多的參數(shù)使用請參考官方文檔。 # user.pyfrom starlette.templating import Jinja2Templatesfrom app import schemas, models
from app.database.database import get_db
from app.home import crud
from fastapi import Depends, HTTPException, Form
from sqlalchemy.orm import Session
from app.models import User
from sqlalchemy.orm import Session
from fastapi import APIRouter, HTTPException,Request
from fastapi.responses import RedirectResponse
userRouter = 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 例子可以發(fā)現(xiàn)，我在上下文 request 中通過判斷路由的請求方式來進行響應(yīng)的邏輯處理,，比如如果不是 Post請求 就把它重定向到 login 頁面等等,。那么就需要同時支持多種請求方式了，巧合的是,，我在 FastAPI 文檔中找不到相應(yīng)的說明,，剛開始的時候我也迷糊了一陣。所以,，只能干源碼了,。
直接進入 APIRouter 類所在的文件，發(fā)現(xiàn)新大陸,。
在 APIRouter 下有個叫 add_api_route 的方法,，支持 http方法 以列表的形式作為參數(shù)傳入，所以就換成了下面這種寫法： 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
初始化數(shù)據(jù)庫文件：
from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 創(chuàng)建數(shù)據(jù)庫連接URI
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:[email protected]:3306/blog"
# 初始化
engine = create_engine(
SQLALCHEMY_DATABASE_URL
)
# 創(chuàng)建DBSession類型
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 創(chuàng)建基類 用于繼承 也可以放到初始化文件中
Base = declarative_base()
# 獲取數(shù)據(jù)庫會話,，用于數(shù)據(jù)庫的各種操作
def get_db():
db = SessionLocal()
數(shù)據(jù)庫模型文件： from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, DateTime, Text
from sqlalchemy.orm import relationship
from datetime import datetime
from flask_login import UserMixin
import uuid
from app.database.database import Base
class 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 "缺少參數(shù)"
刪
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 "缺少參數(shù)"
改 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 "缺少參數(shù)"
查 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,，真正部署的時候可不能這么魯莽哇,，錯誤的捕捉，數(shù)據(jù)庫的回滾,，語句必須嚴謹,。 在路由方法中，有個叫 response_model 的參數(shù),，用于限制路由方法的返回字段,。 官方文檔實例： from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
app = FastAPI()
class UserIn(BaseModel):
username: str
password: str
email: EmailStr
full_name: str = None
class 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 作為請求體參數(shù)傳入，返回時必須滿足 UserOut 模型,。
場景的話,，可以想象用戶登陸時需要傳入用戶名和密碼，用戶登陸成功之后在首頁上展示用戶名的郵件,，不展示密碼,。嗯，這樣就合理了,。 所以在數(shù)據(jù)庫操作的時候,，可以自己定義傳入和返回的模型字段來做有效的限制，你只需要繼承 pydantic 中的 BaseModel 基類即可,，看起來是那么的簡單合理,。 在各種 http資源 不存在或者訪問異常的時候都需要有 http狀態(tài)碼 和 異常說明，例如,， 404 Not Found 錯誤,，Post請求出現(xiàn)的 422,，服務(wù)端的 500 錯誤，所以如何在程序中合理的引發(fā)異常,，就變得格外重要了,。
看看 FastAPI 中如何使用異常處理
from fastapi import FastAPI, HTTPException
app = 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，傳入狀態(tài)碼 和 詳細說明,，在出現(xiàn)邏輯錯誤時拋出異常,。 改寫 HTTPException from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class UnicornException(Exception):
def __init__(self, name: str):
self.name = name
app = 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 類，在出現(xiàn)服務(wù)端錯誤時拋出 418 錯誤,，并附上錯誤說明,。 自定義自己的異常處理代碼 from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
app = 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="開空調(diào)啊")
return {"item_id": item_id}
合理的使用異常處理機制，能讓項目代碼更健壯,，客戶端更友好,，也易于維護。 在茫茫的 FastAPI 文檔中我盡可能摸索出一些易用,，實用，好用的功能來和大家分享,，并嘗試投入到實際的生產(chǎn)環(huán)境中,，在這個過程中去學(xué)習(xí)更多的東西，體驗更好的服務(wù)性能,。 FastAPI 官方文檔十分的龐大,，有非常多的地方還沒有普及和深入，比如 FastAPI 的安全加密,，中間件的使用,，應(yīng)用部署等等。哈,，來日方長 !!! 需要學(xué)習(xí)更多關(guān)于FastAPI 知識的話,，可以戳閱讀全部，獲取詳情： 參考文檔：https://fastapi./tutorial
|
