代码收藏家技术教程 2025-04-04

【Python】Dash与FastAPI前后端集成实践指南

概述

项目中需要快速搭建一个前后端系统，涉及到dash-fastapi架构的时候，对该架构的时候进行总结。本文主要总结的是对该架构的基本使用，后续再对该架构的项目源码进行总结分析

此处实现一个小的demo，迷你任务管理器，后端使用FastAPI，前端则使用Dash，数据存储暂时使用列表进行存储，主要功能如下

任务列表展示: 前端页面显示一个简单的任务列表，包含任务标题和状态。

添加任务: 用户可以在前端输入任务标题，点击按钮添加新任务。

刷新任务列表: 点击按钮可以刷新任务列表，从后端获取最新数据。

整体架构理解

代码主体架构

后端

main.py （Fast API主文件）

requirements.txt（后端依赖）

前端

app.py（Dash主文件）

api_client.py（前端API客户端）

layoputs.py（前端布局）

callbacks.py（前端回调函数）

requirements.txt(后端依赖）

主要逻辑理解

代码中具体体现

后端

main.py：后端，也就类似于厨房。专门负责接收顾客的订单，然后准备食物（构建响应）并告知服务器食物准备后

tasks_db = []：通过列表内存列表，类似于厨师的菜单列表。其记录了餐厅可以提供的菜品，也就是后端可以完成的任务

@app.get ：厨师提供给服务员今日菜单，服务员发送get请求的时候，就可以知道后端提供什么服务（从tasks_db中获取）

@app.post：创创建任务，类似于服务员将菜单传给厨房；后面的逻辑就是请求响应的基本逻辑

Task（使用Pydantic模型）：菜单上的菜品叙述，规定了每个任务包含哪些信息，提供任务名以及状态是否完成

前端

layouts.py：餐厅的菜单，定义了顾客可以看到什么，也就是前端显示的页面

dcc.Input：点餐单的填写区域，顾客要吃什么

dbc.Button：提交按钮，这里可以对应设计供，例如提交点餐单或者是刷新菜单信息

html.Div：上菜的盘子，厨房准备后的食物会放进这个盘子里展示给顾客

callbacks.py：服务员接收到顾客的指令应该如何行动

api_client.py：点餐系统，帮助前端与后端沟通，服务员与厨师之间的沟通

具体实现

该实例主要用于理解该结构的运行

代码

后端：主要提供两个方法，获取所有任务列表和创建新任务

# backend/main.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List

app = FastAPI()

# 模拟内存数据库 (使用 Python 列表)
tasks_db = []
# 用于生成唯一的用户ID
task_id_counter = 1

class Task(BaseModel):
    id: int
    title: str
    status: str = "待完成"  # 默认状态

class TaskCreate(BaseModel):
    title: str

class TaskResponse(BaseModel):
    tasks: List[Task]

@app.get("/api/tasks", response_model=TaskResponse)
async def get_tasks():
    """获取所有任务列表"""
    return TaskResponse(tasks=tasks_db)

@app.post("/api/tasks", response_model=Task)
async def create_task(task_create: TaskCreate):
    """创建新任务"""
    global task_id_counter
    new_task = Task(id=task_id_counter, title=task_create.title)
    tasks_db.append(new_task)
    task_id_counter += 1
    return new_task

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)

后端依赖：requirements.txt

fastapi
uvicorn
pydantic

前端代码：api_client.py（向服务端发起请求）

import requests

API_BASE_URL = "http://localhost:8000/api"  # 后端 API 基础 URL

def get_task_list():
    """获取任务列表"""
    url = f"{API_BASE_URL}/tasks"
    response = requests.get(url)
    response.raise_for_status()  # 检查请求是否成功 (状态码 2xx)
    return response.json()

def create_new_task(title):
    """创建新任务"""
    url = f"{API_BASE_URL}/tasks"
    headers = {'Content-Type': 'application/json'}
    data = {'title': title}
    response = requests.post(url, headers=headers, json=data)
    response.raise_for_status()
    return response.json()

前端回调：callbacks.py，当顾客触碰哪些按钮后与后端交互然后返回的逻辑实现

from dash import Output, Input, State
from .app import app  # 导入 Dash app 实例
from frontend import api_client  # 导入 API 客户端
import dash_html_components as html
import  dash

@app.callback(
    Output("task-list-output", "children"),
    [Input("refresh-tasks-button", "n_clicks"),
     Input("add-task-button", "n_clicks")],
    [State("new-task-title", "value")]
)
def update_task_list(refresh_clicks, add_clicks, new_task_title):
    """更新任务列表显示"""
    triggered_id = [p['prop_id'] for p in dash.callback_context.triggered][0]

    if "add-task-button" in triggered_id:
        if new_task_title:
            api_client.create_new_task(new_task_title) # 调用 API 创建新任务

    tasks_data = api_client.get_task_list() # 调用 API 获取任务列表
    task_items = []
    if tasks_data and tasks_data.get('tasks'): # 检查 tasks_data 和 tasks 键是否存在
        for task in tasks_data['tasks']:
            task_items.append(html.Li(f"{task['title']} - 状态: {task['status']} (ID: {task['id']})"))
    else:
        task_items.append(html.Li("暂无任务"))

    return html.Ul(task_items)

前端页面布局layouts.py

import dash_html_components as html
import dash_core_components as dcc
import dash_bootstrap_components as dbc

layout = dbc.Container([
    html.H1("迷你任务管理器"),
    dbc.Row([
        dbc.Col([
            html.Div("任务标题:"),
            dcc.Input(id="new-task-title", type="text", placeholder="请输入任务标题"),
            dbc.Button("添加任务", id="add-task-button", n_clicks=0, className="mt-2"),
        ]),
    ]),
    html.Hr(className="mt-3"),
    html.H4("任务列表"),
    dbc.Button("刷新任务列表", id="refresh-tasks-button", n_clicks=0, className="mb-2"),
    html.Div(id="task-list-output"), # 用于显示任务列表
])

前端依赖

dash
dash-bootstrap-components
requests

pydantic补充

"""
简单事例
"""

# from pydantic import BaseModel
#
# class User(BaseModel):
#     id: int
#     name: str
#     email: str
#     is_active: bool = True  # 默认值
#
# # 示例数据
# user_data = {
#     'id': 1,
#     'name': 'Alice',
#     'email': 'alice@example.com',
# }
#
# # 使用 Pydantic 模型进行数据验证和解析
# user = User(**user_data)
# print(user)

"""
复杂事例的封装
"""
from pydantic import BaseModel
from typing import List

class Address(BaseModel):
    street: str
    city: str
    zip_code: str

class User(BaseModel):
    id: int
    name: str
    address: Address  # 嵌套模型

# 创建嵌套数据
user_data = {
    'id': 1,
    'name': 'John',
    'address': {
        'street': '123 Main St',
        'city': 'New York',
        'zip_code': '10001',
    }
}

user = User(**user_data)
print(user)

前端回调逻辑

@app.callback(
    Output("task-list-output", "children"),
    [Input("refresh-tasks-button", "n_clicks"),
     Input("add-task-button", "n_clicks")],
    [State("new-task-title", "value")]
)
def update_task_list(refresh_clicks, add_clicks, new_task_title):
    """更新任务列表显示"""
    triggered_id = [p['prop_id'] for p in dash.callback_context.triggered][0]

    if "add-task-button" in triggered_id:
        if new_task_title:
            api_client.create_new_task(new_task_title) # 调用 API 创建新任务

    tasks_data = api_client.get_task_list() # 调用 API 获取任务列表
    task_items = []
    if tasks_data and tasks_data.get('tasks'): # 检查 tasks_data 和 tasks 键是否存在
        for task in tasks_data['tasks']:
            task_items.append(html.Li(f"{task['title']} - 状态: {task['status']} (ID: {task['id']})"))
    else:
        task_items.append(html.Li("暂无任务"))

    return html.Ul(task_items)

回调函数callbacks理解

Dash框架中回调函数是实现交互的关键，其一可以连接前端的UI组件交互和侯丹数据的处理逻辑（调用API或者更新图表操作），从而实现动态更新前端UI，而不需要更新整个页面

Output("task-list-output", "children") (输出)

该处定义了回调函数的输出，其指定了当回调函数执行完毕后，哪个前端组件的哪个属性会被更新

html.Div(id="task-list-output") #  <---  这里定义了 id="task-list-output" 的 Div 组件

这个回调函数执行完成后，将会更新 id 为 "task-list-output" 的 Div 组件的 children 属性。换句话说，回调函数的返回值将会被设置为这个 Div 组件的内容，从而更新任务列表的显示

换句话说，output就是上菜的盘子，盘子里面的内容就是children属性

[Input("refresh-tasks-button", "n_clicks"), Input("add-task-button", "n_clicks")] (输入 – 触发器)

指定了当前前端组件的哪些属性发生变化的时候，会触发这个回调函数执行

dbc.Button("刷新任务列表", id="refresh-tasks-button", ...) # <--- 这里定义了 id="refresh-tasks-button" 的按钮

类似于顾客点击菜价查询，服务员就会去问一下菜价，当顾客点击提交餐单的时候，服务员就会立马去厨房下单

[State("new-task-title", "value")] (状态)

指定哪些前端组件的哪些属性的当前值需要传递给回调函数，但是State组件属性的变化不会触发回调函数执行

可以理解State就是顾客在菜单上书写的菜名

当 update_task_list 回调函数被触发执行时 (因为 "刷新任务列表" 按钮或 "添加任务" 按钮被点击了)，Dash 框架会将 id 为 "new-task-title" 的输入框组件的 value 属性的 "当前值" 作为参数传递给 update_task_list 函数。 注意，输入框内容的变化不会直接触发回调函数，只有当 Input 指定的组件属性变化时才会触发