Dash应用直接使用原生FastAPI后端

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)

对应接口访问结果：