美文网首页Python那些事儿大数据Python自动化运维
(进阶篇)Python web框架FastAPI——一个比Fla

(进阶篇)Python web框架FastAPI——一个比Fla

作者: Python进阶学习交流 | 来源:发表于2020-04-16 22:26 被阅读0次

    前言

    上一篇已经初步了解了 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

    相关文章

      网友评论

        本文标题:(进阶篇)Python web框架FastAPI——一个比Fla

        本文链接:https://www.haomeiwen.com/subject/ggpjvhtx.html