【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:点餐系统,帮助前端与后端沟通,服务员与厨师之间的沟通
- layouts.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
指定的组件属性变化时才会触发