Dash及FastAPI依赖的安装#
该特性从Dash的4.1.0正式版本开始引入,并从4.2.0rc0预发布版本开始进一步优化了相关稳定性问题,因此建议大家使用下面的命令,控制Python环境中Dash的最低版本,顺便完成对FastAPI后端依赖的安装:
pip install "dash>=4.2.0rc0" fastapi查看已安装的Dash和FastAPI信息:
2.2 为Dash应用切换到FastAPI后端#
完成Dash的更新以及相关依赖的安装后,我们先以一个非常简单的Dash应用为例,其关键的代码非常简单,只需要在Dash应用实例化时,设置参数backend="fastapi",当前Dash应用就会自动的在底层使用FastAPI代替默认的Flask:
app1.py
import dash import fastapi import feffery_antd_components as fac from dash import html from feffery_dash_utils.style_utils import style app = dash.Dash(__name__, backend="fastapi") app.layout = html.Div( [ fac.AntdAlert( message="当前Dash应用后端类型:FastAPI", description="Dash版本:{},FastAPI版本:{}".format( dash.__version__, fastapi.__version__, ), showIcon=True, ) ], style=style(padding=50), ) if __name__ == "__main__": app.run(debug=True)通过浏览器开发者工具抓包,也可以看到对应的server变成了FastAPI对应的高性能框架uvicorn:
并且底层切换成FastAPI之后,开发模式下的热重载功能也会转向FastAPI自带的,其在检测到项目代码格式错误后,不会直接终止终端里运行中的应用服务,不像默认的基于Flask的热重载还得手动重新执行命令启动应用,这一点很方便:
2.3 基于FastAPI添加自定义接口#
像上文那样将我们的Dash应用底层更换为FastAPI后,我们就可以基于FastAPI的写法添加额外的自定义接口,对应的Dash应用实例对象,其server属性即为内置的FastAPI实例,根据这一点进行接口开发即可:
2.3.1 基础接口测试#
app2.py
import dash from dash import html app = dash.Dash(__name__, backend="fastapi") app.layout = html.Div() @app.server.get("/fastapi-basic-test") def fastapi_basic_test(): """基础接口测试""" return { "api": "fastapi-basic-test", "status": "success", "message": "Dash+FastAPI=⚡", } if __name__ == "__main__": app.run(debug=True)对应接口访问结果:
2.3.2 异步接口测试#
app3.py
import dash from dash import html app = dash.Dash(__name__, backend="fastapi") @app.server.get("/fastapi-async-test") async def fastapi_async_test(): """异步接口测试""" return { "api": "fastapi-async-test", "status": "success", "message": "Dash+FastAPI=⚡", } app.layout = html.Div() if __name__ == "__main__": app.run(debug=True)对应接口访问结果:
2.3.3 SSE接口测试#
app4.py
import asyncio from datetime import datetime import dash from dash import html from fastapi.sse import EventSourceResponse app = dash.Dash(__name__, backend="fastapi") @app.server.get("/fastapi-sse-test", response_class=EventSourceResponse) async def fastapi_sse_test(): """SSE接口测试""" while True: await asyncio.sleep(1) yield { "timestamp": datetime.now().isoformat(), } app.layout = html.Div() if __name__ == "__main__": app.run(debug=True)对应接口访问结果: