(进阶篇)Python web框架FastAPI——一个比Flask和Tornada更高性能的API 框架

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了(进阶篇)Python web框架FastAPI——一个比Flask和Tornada更高性能的API 框架，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含10554字，纯文字阅读大概需要16分钟。

内容图文

前言

上一篇已经初步了解了 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 文档中我尽可能摸索出一些易用，实用，好用的功能来和大家分享，并尝试投入到实际的生产环境中，在这个过程中去学习更多的东西，体验更好的服务性能。

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

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

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

内容总结

以上是互联网集市为您收集整理的(进阶篇)Python web框架FastAPI——一个比Flask和Tornada更高性能的API 框架全部内容，希望文章能够帮你解决(进阶篇)Python web框架FastAPI——一个比Flask和Tornada更高性能的API 框架所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端