前言
上一篇已经初步了解了 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.pyfromstarlette.templatingimportJinja2Templatesfromappimportschemas,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
初始化数据库文件：
fromsqlalchemyimportcreate_enginefromsqlalchemy.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="nonono!!")
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":"TheFooWrestlers"}
@app.get("/items/{item_id}")
asyncdefread_item(item_id:str):
ifitem_idnotinitems:
raiseHTTPException(status_code=404,detail="Itemnotfound")
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