使用 dash 构建 mvvm 整洁架构应用
整洁架构层次
-
领域层 (Domain Layer)
TaskEntity: 核心业务实体，包含业务规则
TaskRepository: 数据访问抽象接口 -
应用层 (Application Layer)
TaskService: 业务逻辑，协调领域对象完成用例 -
接口适配器层 (Interface Adapters)
InMemoryTaskRepository: 仓储接口的具体实现
MVVM 架构层次
-
Model 层
TaskModel: 管理应用数据状态，处理数据过滤和排序 -
ViewModel 层
TaskViewModel: 处理视图逻辑，数据绑定，通知管理 -
View 层
TaskView: UI 组件和布局，用户交互处理
代码示例
"""
使用 Dash 构建的 MVVM + 整洁架构应用
任务管理系统架构层次：
1. 领域层 (Domain) - 业务实体和规则
2. 应用层 (Application) - 用例和业务逻辑
3. 接口适配器层 (Interface Adapters) - Repository 实现
4. MVVM 层:- Model: 数据模型- ViewModel: 业务逻辑和状态管理- View: UI 组件和布局
"""import dash
from dash import dcc, html, Input, Output, State, callback_context, dash_table
import dash_bootstrap_components as dbc
from datetime import datetime
from abc import ABC, abstractmethod
from typing import List, Optional, Dict, Any
import uuid
import json# ======================
# 整洁架构 - 领域层 (Domain Layer)
# ======================class TaskEntity:"""任务实体 - 核心业务对象"""def __init__(self, id: str, title: str, description: str, status: str = "pending", created_at: str = None,priority: str = "medium", due_date: str = None):self.id = idself.title = titleself.description = descriptionself.status = status # "pending" or "completed"self.priority = priority # "low", "medium", "high"self.due_date = due_dateself.created_at = created_at or datetime.now().strftime("%Y-%m-%d %H:%M:%S")def mark_completed(self):"""标记任务为已完成"""self.status = "completed"def mark_pending(self):"""标记任务为待完成"""self.status = "pending"def toggle_status(self):"""切换任务状态"""if self.status == "pending":self.mark_completed()else:self.mark_pending()def is_completed(self) -> bool:"""检查任务是否已完成"""return self.status == "completed"def is_overdue(self) -> bool:"""检查任务是否过期"""if not self.due_date:return Falsereturn datetime.now() > datetime.strptime(self.due_date, "%Y-%m-%d")def to_dict(self) -> Dict[str, Any]:"""转换为字典"""return {'id': self.id,'title': self.title,'description': self.description,'status': self.status,'priority': self.priority,'due_date': self.due_date,'created_at': self.created_at}@classmethoddef from_dict(cls, data: Dict[str, Any]) -> 'TaskEntity':"""从字典创建任务"""return cls(id=data['id'],title=data['title'],description=data['description'],status=data.get('status', 'pending'),priority=data.get('priority', 'medium'),due_date=data.get('due_date'),created_at=data.get('created_at'))class TaskRepository(ABC):"""任务仓储接口 - 数据访问抽象"""@abstractmethoddef add(self, task: TaskEntity) -> None:pass@abstractmethoddef get_by_id(self, task_id: str) -> Optional[TaskEntity]:pass@abstractmethoddef get_all(self) -> List[TaskEntity]:pass@abstractmethoddef update(self, task: TaskEntity) -> None:pass@abstractmethoddef delete(self, task_id: str) -> bool:pass@abstractmethoddef get_by_status(self, status: str) -> List[TaskEntity]:pass@abstractmethoddef get_by_priority(self, priority: str) -> List[TaskEntity]:pass# ======================
# 整洁架构 - 应用层 (Application Layer)
# ======================class TaskService:"""任务服务 - 业务逻辑"""def __init__(self, task_repository: TaskRepository):self.task_repository = task_repositorydef create_task(self, title: str, description: str, priority: str = "medium", due_date: str = None) -> TaskEntity:"""创建新任务"""if not title or not description:raise ValueError("标题和描述不能为空")if priority not in ["low", "medium", "high"]:raise ValueError("优先级必须是 low、medium 或 high")task_id = str(uuid.uuid4())task = TaskEntity(id=task_id, title=title, description=description,priority=priority,due_date=due_date)self.task_repository.add(task)return taskdef get_task(self, task_id: str) -> Optional[TaskEntity]:"""获取任务"""return self.task_repository.get_by_id(task_id)def get_all_tasks(self) -> List[TaskEntity]:"""获取所有任务"""return self.task_repository.get_all()def get_tasks_by_status(self, status: str) -> List[TaskEntity]:"""根据状态获取任务"""return self.task_repository.get_by_status(status)def get_tasks_by_priority(self, priority: str) -> List[TaskEntity]:"""根据优先级获取任务"""return self.task_repository.get_by_priority(priority)def toggle_task_status(self, task_id: str) -> Optional[TaskEntity]:"""切换任务状态"""task = self.task_repository.get_by_id(task_id)if task:task.toggle_status()self.task_repository.update(task)return taskreturn Nonedef update_task_priority(self, task_id: str, priority: str) -> Optional[TaskEntity]:"""更新任务优先级"""if priority not in ["low", "medium", "high"]:raise ValueError("优先级必须是 low、medium 或 high")task = self.task_repository.get_by_id(task_id)if task:task.priority = priorityself.task_repository.update(task)return taskreturn Nonedef delete_task(self, task_id: str) -> bool:"""删除任务"""return self.task_repository.delete(task_id)def get_task_statistics(self) -> Dict[str, Any]:"""获取任务统计"""tasks = self.task_repository.get_all()total = len(tasks)completed = len([task for task in tasks if task.is_completed()])pending = total - completed# 按优先级统计low_priority = len([task for task in tasks if task.priority == "low"])medium_priority = len([task for task in tasks if task.priority == "medium"])high_priority = len([task for task in tasks if task.priority == "high"])# 过期任务overdue = len([task for task in tasks if task.is_overdue() and not task.is_completed()])return {'total': total,'completed': completed,'pending': pending,'low_priority': low_priority,'medium_priority': medium_priority,'high_priority': high_priority,'overdue': overdue,'completion_rate': round(completed / total * 100, 2) if total > 0 else 0}# ======================
# 整洁架构 - 接口适配器层 (Interface Adapters)
# ======================class InMemoryTaskRepository(TaskRepository):"""内存任务仓储实现"""def __init__(self):self.tasks: Dict[str, TaskEntity] = {}def add(self, task: TaskEntity) -> None:self.tasks[task.id] = taskdef get_by_id(self, task_id: str) -> Optional[TaskEntity]:return self.tasks.get(task_id)def get_all(self) -> List[TaskEntity]:return list(self.tasks.values())def update(self, task: TaskEntity) -> None:if task.id in self.tasks:self.tasks[task.id] = taskdef delete(self, task_id: str) -> bool:if task_id in self.tasks:del self.tasks[task_id]return Truereturn Falsedef get_by_status(self, status: str) -> List[TaskEntity]:return [task for task in self.tasks.values() if task.status == status]def get_by_priority(self, priority: str) -> List[TaskEntity]:return [task for task in self.tasks.values() if task.priority == priority]# ======================
# MVVM 架构 - Model 层
# ======================class TaskModel:"""任务模型 - 管理应用数据状态"""def __init__(self, task_service: TaskService):self.task_service = task_serviceself._tasks = []self._filtered_tasks = []self._statistics = {}self._current_filter = "all"self._current_sort = "created"self._load_data()def _load_data(self):"""加载数据"""self._tasks = self.task_service.get_all_tasks()self._apply_filters()self._update_statistics()def _apply_filters(self):"""应用当前过滤器"""if self._current_filter == "all":self._filtered_tasks = self._tasks.copy()elif self._current_filter == "pending":self._filtered_tasks = [task for task in self._tasks if task.status == "pending"]elif self._current_filter == "completed":self._filtered_tasks = [task for task in self._tasks if task.status == "completed"]elif self._current_filter.startswith("priority_"):priority = self._current_filter.split("_")[1]self._filtered_tasks = [task for task in self._tasks if task.priority == priority]self._apply_sorting()def _apply_sorting(self):"""应用排序"""if self._current_sort == "created":self._filtered_tasks.sort(key=lambda x: x.created_at, reverse=True)elif self._current_sort == "priority":priority_order = {"high": 3, "medium": 2, "low": 1}self._filtered_tasks.sort(key=lambda x: priority_order[x.priority], reverse=True)elif self._current_sort == "due_date":self._filtered_tasks.sort(key=lambda x: x.due_date or "9999-99-99")def _update_statistics(self):"""更新统计信息"""self._statistics = self.task_service.get_task_statistics()@propertydef tasks(self) -> List[Dict[str, Any]]:"""获取任务列表（用于视图）"""return [task.to_dict() for task in self._filtered_tasks]@propertydef statistics(self) -> Dict[str, Any]:"""获取统计信息（用于视图）"""return self._statisticsdef add_task(self, title: str, description: str, priority: str = "medium", due_date: str = None):"""添加任务"""task = self.task_service.create_task(title, description, priority, due_date)self._tasks.append(task)self._apply_filters()self._update_statistics()return taskdef toggle_task_status(self, task_id: str):"""切换任务状态"""task = self.task_service.toggle_task_status(task_id)if task:# 更新本地数据for i, t in enumerate(self._tasks):if t.id == task_id:self._tasks[i] = taskbreakself._apply_filters()self._update_statistics()return taskdef update_task_priority(self, task_id: str, priority: str):"""更新任务优先级"""task = self.task_service.update_task_priority(task_id, priority)if task:# 更新本地数据for i, t in enumerate(self._tasks):if t.id == task_id:self._tasks[i] = taskbreakself._apply_filters()self._update_statistics()return taskdef delete_task(self, task_id: str):"""删除任务"""success = self.task_service.delete_task(task_id)if success:# 更新本地数据self._tasks = [task for task in self._tasks if task.id != task_id]self._apply_filters()self._update_statistics()return successdef set_filter(self, filter_type: str):"""设置过滤器"""self._current_filter = filter_typeself._apply_filters()def set_sort(self, sort_type: str):"""设置排序"""self._current_sort = sort_typeself._apply_filters()# ======================
# MVVM 架构 - ViewModel 层
# ======================class TaskViewModel:"""任务视图模型 - 处理视图逻辑和数据绑定"""def __init__(self, model: TaskModel):self.model = modelself._notification = None@propertydef tasks(self) -> List[Dict[str, Any]]:"""获取任务数据"""return self.model.tasks@propertydef statistics(self) -> Dict[str, Any]:"""获取统计信息"""return self.model.statistics@propertydef notification(self) -> Optional[Dict[str, Any]]:"""获取通知信息"""return self._notificationdef add_task(self, title: str, description: str, priority: str = "medium", due_date: str = None):"""添加任务"""try:task = self.model.add_task(title, description, priority, due_date)self._notification = {"message": f"任务 '{title}' 已创建","type": "success"}return Trueexcept Exception as e:self._notification = {"message": f"创建任务失败: {str(e)}","type": "error"}return Falsedef toggle_task_status(self, task_id: str):"""切换任务状态"""task = self.model.toggle_task_status(task_id)if task:status = "已完成" if task.is_completed() else "待完成"self._notification = {"message": f"任务状态已更新为 '{status}'","type": "success"}return Trueelse:self._notification = {"message": "任务未找到","type": "error"}return Falsedef update_task_priority(self, task_id: str, priority: str):"""更新任务优先级"""try:task = self.model.update_task_priority(task_id, priority)if task:self._notification = {"message": f"任务优先级已更新为 '{priority}'","type": "success"}return Trueelse:self._notification = {"message": "任务未找到","type": "error"}return Falseexcept Exception as e:self._notification = {"message": f"更新优先级失败: {str(e)}","type": "error"}return Falsedef delete_task(self, task_id: str):"""删除任务"""success = self.model.delete_task(task_id)if success:self._notification = {"message": "任务已删除","type": "success"}else:self._notification = {"message": "任务未找到","type": "error"}return successdef set_filter(self, filter_type: str):"""设置过滤器"""self.model.set_filter(filter_type)def set_sort(self, sort_type: str):"""设置排序"""self.model.set_sort(sort_type)def clear_notification(self):"""清除通知"""self._notification = None# ======================
# MVVM 架构 - View 层
# ======================class TaskView:"""任务视图 - 负责UI呈现和用户交互"""def __init__(self, viewmodel: TaskViewModel):self.viewmodel = viewmodelself.app = dash.Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])self.setup_layout()self.setup_callbacks()def setup_layout(self):"""设置应用布局"""self.app.layout = dbc.Container([# 存储组件dcc.Store(id='notification-trigger', data=0),dcc.Interval(id='auto-refresh', interval=30000, n_intervals=0), # 30秒自动刷新# 标题区域dbc.Row([dbc.Col([html.H1("任务管理系统 - MVVM + 整洁架构", className="text-center my-4 text-primary"),html.P("结合 MVVM 模式和整洁架构的 Dash 应用示例", className="text-center text-muted mb-4")])]),# 通知区域dbc.Row([dbc.Col([html.Div(id="notification-area")])]),# 统计信息区域dbc.Row([dbc.Col([dbc.Card([dbc.CardBody([dbc.Row([dbc.Col([self._create_stat_card("总任务数", "total-tasks", "primary")], width=2),dbc.Col([self._create_stat_card("待完成", "pending-tasks", "warning")], width=2),dbc.Col([self._create_stat_card("已完成", "completed-tasks", "success")], width=2),dbc.Col([self._create_stat_card("完成率", "completion-rate", "info")], width=2),dbc.Col([self._create_stat_card("高优先级", "high-priority-tasks", "danger")], width=2),dbc.Col([self._create_stat_card("已过期", "overdue-tasks", "dark")], width=2),])])], className="mb-4")])]),# 控制面板dbc.Row([dbc.Col([dbc.Card([dbc.CardBody([dbc.Row([dbc.Col([dbc.ButtonGroup([dbc.Button("全部任务", id="filter-all", color="primary", outline=True),dbc.Button("待完成", id="filter-pending", color="warning", outline=True),dbc.Button("已完成", id="filter-completed", color="success", outline=True),], className="me-2"),dbc.ButtonGroup([dbc.Button("高优先级", id="filter-high", color="danger", outline=True),dbc.Button("中优先级", id="filter-medium", color="info", outline=True),dbc.Button("低优先级", id="filter-low", color="secondary", outline=True),]),], width=8),dbc.Col([dbc.Select(id="sort-select",options=[{"label": "按创建时间", "value": "created"},{"label": "按优先级", "value": "priority"},{"label": "按截止日期", "value": "due_date"},],value="created",className="float-end")], width=4)])])], className="mb-4")])]),# 添加任务区域dbc.Row([dbc.Col([dbc.Card([dbc.CardHeader("添加新任务"),dbc.CardBody([dbc.Row([dbc.Col([dbc.Input(id="task-title-input",placeholder="任务标题",type="text",className="mb-2")], width=4),dbc.Col([dbc.Input(id="task-desc-input",placeholder="任务描述",type="text",className="mb-2")], width=4),dbc.Col([dbc.Select(id="task-priority-select",options=[{"label": "低优先级", "value": "low"},{"label": "中优先级", "value": "medium"},{"label": "高优先级", "value": "high"},],value="medium",className="mb-2")], width=2),dbc.Col([dbc.Input(id="task-due-date",type="date",className="mb-2")], width=2),]),dbc.Row([dbc.Col([dbc.Button("添加任务", id="add-task-btn", color="primary",className="w-100")])])])], className="mb-4")])]),# 任务列表区域dbc.Row([dbc.Col([html.Div(id="tasks-container")])])], fluid=True)def _create_stat_card(self, title: str, element_id: str, color: str):"""创建统计卡片"""return html.Div([html.H6(title, className="card-title"),html.H3(id=element_id, className=f"card-text text-{color}")], className="text-center")def setup_callbacks(self):"""设置应用回调"""# 更新统计信息@self.app.callback([Output("total-tasks", "children"),Output("pending-tasks", "children"),Output("completed-tasks", "children"),Output("completion-rate", "children"),Output("high-priority-tasks", "children"),Output("overdue-tasks", "children")],[Input("filter-all", "n_clicks"),Input("filter-pending", "n_clicks"),Input("filter-completed", "n_clicks"),Input("filter-high", "n_clicks"),Input("filter-medium", "n_clicks"),Input("filter-low", "n_clicks"),Input("sort-select", "value"),Input("auto-refresh", "n_intervals")])def update_statistics(all_clicks, pending_clicks, completed_clicks, high_clicks, medium_clicks, low_clicks, sort_value, n_intervals):"""更新统计信息"""ctx = callback_contextif ctx.triggered:trigger_id = ctx.triggered[0]['prop_id'].split('.')[0]# 处理过滤器if trigger_id == "filter-all":self.viewmodel.set_filter("all")elif trigger_id == "filter-pending":self.viewmodel.set_filter("pending")elif trigger_id == "filter-completed":self.viewmodel.set_filter("completed")elif trigger_id == "filter-high":self.viewmodel.set_filter("priority_high")elif trigger_id == "filter-medium":self.viewmodel.set_filter("priority_medium")elif trigger_id == "filter-low":self.viewmodel.set_filter("priority_low")# 处理排序if trigger_id == "sort-select":self.viewmodel.set_sort(sort_value)stats = self.viewmodel.statisticstotal = stats['total']pending = stats['pending']completed = stats['completed']completion_rate = f"{stats['completion_rate']}%"high_priority = stats['high_priority']overdue = stats['overdue']return total, pending, completed, completion_rate, high_priority, overdue# 添加任务@self.app.callback([Output("notification-trigger", "data"),Output("task-title-input", "value"),Output("task-desc-input", "value")],[Input("add-task-btn", "n_clicks")],[State("task-title-input", "value"),State("task-desc-input", "value"),State("task-priority-select", "value"),State("task-due-date", "value")])def add_task(n_clicks, title, description, priority, due_date):"""添加新任务"""if n_clicks is None or n_clicks == 0:return 0, "", ""if not title or not description:return 0, title, descriptionsuccess = self.viewmodel.add_task(title, description, priority, due_date)if success:return 1, "", ""else:return 0, title, description# 显示通知@self.app.callback(Output("notification-area", "children"),[Input("notification-trigger", "data")])def show_notification(trigger):"""显示通知"""notification = self.viewmodel.notificationif notification:color = "success" if notification["type"] == "success" else "danger"alert = dbc.Alert(notification["message"],color=color,dismissable=True,is_open=True)# 清除通知self.viewmodel.clear_notification()return alertreturn ""# 渲染任务列表@self.app.callback(Output("tasks-container", "children"),[Input("filter-all", "n_clicks"),Input("filter-pending", "n_clicks"),Input("filter-completed", "n_clicks"),Input("filter-high", "n_clicks"),Input("filter-medium", "n_clicks"),Input("filter-low", "n_clicks"),Input("sort-select", "value"),Input("auto-refresh", "n_intervals")])def render_tasks(all_clicks, pending_clicks, completed_clicks, high_clicks, medium_clicks, low_clicks, sort_value, n_intervals):"""渲染任务列表"""tasks = self.viewmodel.tasksif not tasks:return dbc.Alert("暂无任务", color="info")task_cards = []for task in tasks:# 状态徽章status_color = "success" if task['status'] == 'completed' else "warning"status_text = "已完成" if task['status'] == 'completed' else "待完成"# 优先级徽章priority_color = {"low": "secondary","medium": "info", "high": "danger"}.get(task['priority'], "secondary")priority_text = {"low": "低优先级","medium": "中优先级","high": "高优先级"}.get(task['priority'], "未知")# 过期提醒due_date_text = ""due_date_color = ""if task['due_date']:due_date = datetime.strptime(task['due_date'], "%Y-%m-%d")if due_date < datetime.now() and task['status'] != 'completed':due_date_text = f" (已过期)"due_date_color = "text-danger"else:due_date_text = f" ({task['due_date']})"card = dbc.Card([dbc.CardBody([dbc.Row([dbc.Col([html.H5(task['title']),html.P(task['description'], className="text-muted"),html.Div([html.Small(f"创建时间: {task['created_at']}", className="text-muted me-3"),html.Small(f"截止日期: {task['due_date'] or '无'}{due_date_text}", className=f"text-muted {due_date_color}")])], width=8),dbc.Col([dbc.Badge(status_text, color=status_color, className="me-1 mb-1"),dbc.Badge(priority_text, color=priority_color, className="mb-1"),html.Div([dbc.Button("切换状态", id={"type": "toggle-btn", "index": task['id']},color="primary",size="sm",className="me-1 mb-1"),dbc.DropdownMenu(label="优先级",children=[dbc.DropdownMenuItem("高优先级", id={"type": "priority-btn", "index": task['id'], "value": "high"}),dbc.DropdownMenuItem("中优先级", id={"type": "priority-btn", "index": task['id'], "value": "medium"}),dbc.DropdownMenuItem("低优先级", id={"type": "priority-btn", "index": task['id'], "value": "low"}),],color="info",size="sm",className="me-1 mb-1 d-inline-block"),dbc.Button("删除", id={"type": "delete-btn", "index": task['id']},color="danger",size="sm",className="mb-1")])], width=4, className="text-end")])])], className="mb-3")task_cards.append(card)return task_cards# 处理任务操作（切换状态、更新优先级、删除）@self.app.callback(Output("notification-trigger", "data", allow_duplicate=True),[Input({"type": "toggle-btn", "index": dash.dependencies.ALL}, "n_clicks"),Input({"type": "priority-btn", "index": dash.dependencies.ALL}, "n_clicks"),Input({"type": "delete-btn", "index": dash.dependencies.ALL}, "n_clicks")],prevent_initial_call=True)def handle_task_actions(toggle_clicks, priority_clicks, delete_clicks):"""处理任务操作"""ctx = callback_contextif not ctx.triggered:return 0trigger = ctx.triggered[0]trigger_id = trigger['prop_id']trigger_value = trigger['value'] if 'value' in trigger else None# 解析触发按钮的类型和IDif 'toggle-btn' in trigger_id:task_id = eval(trigger_id.split('.')[0])['index']self.viewmodel.toggle_task_status(task_id)return 1elif 'priority-btn' in trigger_id:task_id = eval(trigger_id.split('.')[0])['index']priority = eval(trigger_id.split('.')[0])['value']self.viewmodel.update_task_priority(task_id, priority)return 1elif 'delete-btn' in trigger_id:task_id = eval(trigger_id.split('.')[0])['index']self.viewmodel.delete_task(task_id)return 1return 0def run(self, debug=True):"""运行应用"""self.app.run_server(debug=debug)# ======================
# 应用启动和依赖注入
# ======================def create_app():"""创建应用实例 - 依赖注入容器"""# 创建仓储 (整洁架构 - 接口适配器层)task_repository = InMemoryTaskRepository()# 创建服务 (整洁架构 - 应用层)task_service = TaskService(task_repository)# 创建模型 (MVVM - Model层)task_model = TaskModel(task_service)# 创建视图模型 (MVVM - ViewModel层)task_viewmodel = TaskViewModel(task_model)# 创建视图 (MVVM - View层)app = TaskView(task_viewmodel)return appif __name__ == "__main__":# 创建并运行应用app = create_app()app.run(debug=True)