FastAPI + Dash

FastAPI + Dash#

首先定义 Dash 子应用:

!pip install -q fastapi uvicorn dash
# dashapp.py
import dash
from dash.dependencies import Input, Output
from dash import dcc, html
import pandas as pd

def create_dash_app(requests_pathname_prefix: str = None) -> dash.Dash:
    df = pd.read_csv('https://raw.githubusercontent.com/plotly/datasets/master/hello-world-stock.csv')

    app = dash.Dash(__name__, requests_pathname_prefix=requests_pathname_prefix)
    app.scripts.config.serve_locally = False
    dcc._js_dist[0]['external_url'] = 'https://cdn.plot.ly/plotly-basic-latest.min.js'

    app.layout = html.Div([
        html.H1('Stock Tickers'),
        dcc.Dropdown(
            id='my-dropdown',
            options=[
                {'label': 'Tesla', 'value': 'TSLA'},
                {'label': 'Apple', 'value': 'AAPL'},
                {'label': 'Coke', 'value': 'COKE'}
            ],
            value='TSLA'
        ),
        dcc.Graph(id='my-graph')
    ], className="container")

    @app.callback(Output('my-graph', 'figure'),
                  [Input('my-dropdown', 'value')])
    def update_graph(selected_dropdown_value):
        dff = df[df['Stock'] == selected_dropdown_value]
        return {
            'data': [{
                'x': dff.Date,
                'y': dff.Close,
                'line': {
                    'width': 3,
                    'shape': 'spline'
                }
            }],
            'layout': {
                'margin': {
                    'l': 30,
                    'r': 20,
                    'b': 30,
                    't': 20
                }
            }
        }
    return app

主应用:

from fastapi import FastAPI, Request
from fastapi.middleware.wsgi import WSGIMiddleware


app = FastAPI()

@app.get("/")
def read_main():
    return {
        "routes": [
            {"method": "GET", "path": "/", "summary": "Landing"},
            {"method": "GET", "path": "/status", "summary": "App status"},
            {"method": "GET", "path": "/dash",
                "summary": "Sub-mounted Dash application"},
        ]
    }

@app.get("/status")
def get_status():
    return {"status": "ok"}


dash_app = create_dash_app(requests_pathname_prefix="/dash/")
app.mount("/dash", WSGIMiddleware(dash_app.server))
if __name__ == "__main__":
    import asyncio
    import uvicorn
    config = uvicorn.Config(app, port=3004, reload=False,)
    server = uvicorn.Server(config)
    loop = asyncio.get_event_loop()
    loop.create_task(server.serve())