（进阶篇）Python web框架FastAPI——一个比Fla

前言

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

蓝图

事实上，FastAPI 并没有关于蓝图 (Blueprint) 的定义，在 FastAPI 中使用 Include_route 方法来添加路由，也就是我们所熟知的蓝图了。

importtime

fromtypingimportList

fromstarlette.templatingimportJinja2Templates

fromfastapiimportDepends, FastAPI, HTTPException

fromstarlette.staticfilesimportStaticFiles

fromstarlette.templatingimportJinja2Templates

fromappimportmodels

fromapp.database.databaseimportSessionLocal, engine

fromapp.homeimportuser,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 Jinja2Templatesfromappimportschemas, models

fromapp.database.databaseimportget_db

fromapp.homeimportcrud

fromfastapiimportDepends, HTTPException, Form

fromsqlalchemy.ormimportSession

fromapp.modelsimportUser

fromsqlalchemy.ormimportSession

fromfastapiimportAPIRouter, HTTPException,Request

fromfastapi.responsesimportRedirectResponse

userRouter = APIRouter()

templates = Jinja2Templates(directory="app/templates")# 模板目录

@userRouter.post("/login/", response_model=schemas.UserOut)

asyncdeflogin(*,request: Request,db: Session = Depends(get_db), username: str = Form(None), password: str = Form(None),):

ifrequest.method =="POST":

db_user = db.query(models.User).filter(User.username == username).first()

ifnotdb_user:

raiseHTTPException(status_code=400, detail="用户不存在")

print("验证通过 ！！！")

returnRedirectResponse('/index')

returntemplates.TemplateResponse("user/login.html", {"request": request})

看起来比 Flask 添加蓝图要轻松许多。

同时支持多种请求方式

在上面的 login 例子可以发现，我在上下文 request 中通过判断路由的请求方式来进行响应的逻辑处理，比如如果不是 Post请求 就把它重定向到 login 页面等等。那么就需要同时支持多种请求方式了，巧合的是，我在 FastAPI 文档中找不到相应的说明，刚开始的时候我也迷糊了一阵。所以，只能干源码了。

直接进入 APIRouter 类所在的文件，发现新大陆。

在APIRouter 下有个叫 add_api_route 的方法，支持 http方法 以列表的形式作为参数传入，所以就换成了下面这种写法：

asyncdeflogin(*,request: Request,db: Session = Depends(get_db), username: str = Form(None), password: str = Form(None),):

ifrequest.method =="POST":

db_user = db.query(models.User).filter(User.username == username).first()

ifnotdb_user:

raiseHTTPException(status_code=400, detail="用户不存在")

print("验证通过 ！！！")

returnRedirectResponse('/index')

returntemplates.TemplateResponse("user/login.html", {"request": request})

asyncdefuserList(*,request: Request,db: Session = Depends(get_db)):

userList = db.query(models.User).all()

returntemplates.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_engine

fromsqlalchemy.ext.declarativeimportdeclarative_base

fromsqlalchemy.ormimportsessionmaker

# 创建数据库连接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()

# 获取数据库会话，用于数据库的各种操作

defget_db():

db = SessionLocal()

    数据库模型文件：

fromsqlalchemyimportBoolean, Column, ForeignKey, Integer, String, DateTime, Text

fromsqlalchemy.ormimportrelationship

fromdatetimeimportdatetime

fromflask_loginimportUserMixin

importuuid

fromapp.database.databaseimportBase

classUser(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

# 文章表

classArticle(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

增

asyncdefarticleDetailAdd(*,request: Request,db: Session = Depends(get_db),d_content:str,uid:int):

ifrequest.method =="POST":

addArticleDetail= Article_Detail(d_content=d_content,uid=uid)

db.add(addArticleDetail)

db.commit()

db.refresh(addArticleDetail)

print("添加成功 ！！！")

return"添加成功"

return"缺少参数"

删

asyncdefarticleDetailDel(*,request: Request,db: Session = Depends(get_db),aid:int):

ifrequest.method =="POST":

db.query(Article_Detail).filter(Article_Detail.id == aid).delete()

db.commit()

print("删除成功 ！！！")

return"删除成功"

return"缺少参数"

改

asyncdefarticleDetailUpdate(*,request: Request,db: Session = Depends(get_db),aid:int,d_content:str):

ifrequest.method =="POST":

articleInfo= db.query(Article_Detail).filter(Article_Detail.id == aid).first()

print(articleInfo)

ifnotarticleInfo:

raiseHTTPException(status_code=400, detail="no no no !!")

articleInfo.d_content = d_content

db.commit()

print("提交成功 ！！！")

return"更新成功"

return"缺少参数"

查

asyncdefarticleDetailIndex(*,request: Request,db: Session = Depends(get_db),):

articleDetailList = db.query(models.Article_Detail).all()

returntemplates.TemplateResponse("articleDetail/articleDetail-index.html", {"request": request,"articleDetailList":articleDetailList})

这里是一些示例的 crud，真正部署的时候可不能这么鲁莽哇，错误的捕捉，数据库的回滚，语句必须严谨。

数据验证

在路由方法中，有个叫 response_model 的参数，用于限制路由方法的返回字段。

官方文档实例：

fromfastapiimportFastAPI

frompydanticimportBaseModel, EmailStr

app = FastAPI()

classUserIn(BaseModel):

username: str

password: str

email: EmailStr

full_name: str =None

classUserOut(BaseModel):

username: str

email: EmailStr

full_name: str =None

@app.post("/user/", response_model=UserOut)

asyncdefcreate_user(*, user: UserIn):

returnuser

意思是 UserIn 作为请求体参数传入，返回时必须满足 UserOut 模型。

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

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

异常处理

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

看看 FastAPI 中如何使用异常处理

fromfastapiimportFastAPI, HTTPException

app = FastAPI()

items = {"foo":"The Foo Wrestlers"}

@app.get("/items/{item_id}")

asyncdefread_item(item_id: str):

ifitem_idnotinitems:

raiseHTTPException(status_code=404, detail="Item not found")

return{"item": items[item_id]}

使用 HTTPException，传入状态码 和 详细说明，在出现逻辑错误时抛出异常。

改写HTTPException

fromfastapiimportFastAPI, Request

fromfastapi.responsesimportJSONResponse

classUnicornException(Exception):

def__init__(self, name: str):

self.name = name

app = FastAPI()

@app.exception_handler(UnicornException)

asyncdefunicorn_exception_handler(request: Request, exc: UnicornException):

returnJSONResponse(

status_code=418,

content={"message":f"我家热得快炸了..."},

)

@app.get("/unicorns/{name}")

asyncdefread_unicorn(name: str):

ifname =="yolo":

raiseUnicornException(name=name)

return{"unicorn_name": name}

UnicornException 继承自 Python 自带的 Exception 类，在出现服务端错误时抛出 418 错误，并附上错误说明。

    自定义自己的异常处理代码

fromfastapiimportFastAPI, HTTPException

fromfastapi.exceptionsimportRequestValidationError

fromfastapi.responsesimportPlainTextResponse

fromstarlette.exceptionsimportHTTPExceptionasStarletteHTTPException

app = FastAPI()

@app.exception_handler(StarletteHTTPException)

asyncdefhttp_exception_handler(request, exc):

returnPlainTextResponse(str(exc.detail), status_code=exc.status_code)

@app.exception_handler(RequestValidationError)

asyncdefvalidation_exception_handler(request, exc):

returnPlainTextResponse(str(exc), status_code=400)

@app.get("/items/{item_id}")

asyncdefread_item(item_id: int):

ifitem_id ==3:

raiseHTTPException(status_code=418, detail="开空调啊")

return{"item_id": item_id}

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

还有吗？

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

FastAPI 官方文档十分的庞大，有非常多的地方还没有普及和深入，比如 FastAPI 的安全加密，中间件的使用，应用部署等等。哈，来日方长 !!!

    需要学习更多关于FastAPI 知识的话，可以戳阅读全部，获取详情：

    参考文档：https://fastapi.tiangolo.com/tutorial