参考：https://www.cnblogs.com/dcpeng/p/12716560.html

前言

 

上一篇已经初步了解了 FastAPI 的基本使用，但是如果想要真正把 FastAPI 部署上线到服务器，那么你需要了解更多，学习更多。所以本篇内容将注重于 FastAPI 的项目生产环境，诸如 数据库，路由蓝图，数据验证等问题在 FastAPI 中的具体操作和一些自己碰到的坑，分享给正在进攻 FastAPI 的各位小伙伴。

 

 

蓝图

 

事实上，FastAPI 并没有关于蓝图 (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") # 挂载静态文件，指定目录
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, 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 例子可以发现，我在上下文 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_base
from sqlalchemy.orm import sessionmaker

# 创建数据库连接URI
SQLALCHEMY_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, 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 "缺少参数"

 

删

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 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 作为请求体参数传入，返回时必须满足 UserOut 模型。

场景的话，可以想象用户登陆时需要传入用户名和密码，用户登陆成功之后在首页上展示用户名的邮件，不展示密码。嗯，这样就合理了。

 

所以在数据库操作的时候，可以自己定义传入和返回的模型字段来做有效的限制，你只需要继承 pydantic 中的 BaseModel 基类即可，看起来是那么的简单合理。

 

异常处理

 

在各种 http资源 不存在或者访问异常的时候都需要有 http状态码 和 异常说明，例如， 404 Not Found 错误，Post请求出现的 422，服务端的 500 错误，所以如何在程序中合理的引发异常，就变得格外重要了。

 

看看 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，传入状态码 和 详细说明，在出现逻辑错误时抛出异常。

 

改写 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 类，在出现服务端错误时抛出 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="开空调啊")
    return {"item_id": item_id}

 

合理的使用异常处理机制，能让项目代码更健壮，客户端更友好，也易于维护。

 

还有吗？

 

在茫茫的 FastAPI 文档中我尽可能摸索出一些易用，实用，好用的功能来和大家分享，并尝试投入到实际的生产环境中，在这个过程中去学习更多的东西，体验更好的服务性能。