使用 dash 构建整洁架构应用

整洁架构说明

这个应用严格遵循了整洁架构的原则：

1. 领域层 (Domain Layer)
   Task: 核心业务实体，包含业务规则和方法
   TaskRepository: 数据访问抽象接口
   TaskFilter: 值对象，封装过滤逻辑

2. 应用层 (Application Layer)
   TaskService: 业务逻辑，协调领域对象完成用例
   不依赖任何外部框架或技术细节

3. 接口适配器层 (Interface Adapters)
   InMemoryTaskRepository: 仓储接口的具体实现
   TaskPresenter: 数据展示器，准备视图所需的数据格式
   TaskController: 处理用户输入，调用应用层服务

4. 框架层 (Frameworks & Drivers)
   TaskManagerApp: Dash 应用，处理 UI 渲染和用户交互
   依赖关系从外层指向内层

依赖关系规则

内层不依赖于外层
依赖方向：框架层 → 接口适配器层 → 应用层 → 领域层
所有跨层通信都通过接口进行

代码示例

"""
使用 Dash 构建的整洁架构应用示例
任务管理系统架构层次：
1. 领域层 (Domain) - 业务实体和规则
2. 应用层 (Application) - 用例和业务逻辑
3. 接口适配器层 (Interface Adapters) - 控制器、Presenter、Repository 实现
4. 框架层 (Frameworks & Drivers) - UI、数据库、外部服务
"""import dash
from dash import dcc, html, Input, Output, State, callback_context
import dash_bootstrap_components as dbc
from datetime import datetime
from abc import ABC, abstractmethod
from typing import List, Optional, Dict, Any
import uuid# ======================
# 领域层 (Domain Layer)
# ======================class Task:"""任务实体 - 核心业务对象"""def __init__(self, id: str, title: str, description: str, status: str = "pending", created_at: str = None):self.id = idself.title = titleself.description = descriptionself.status = status  # "pending" or "completed"self.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 to_dict(self) -> Dict[str, Any]:"""转换为字典"""return {'id': self.id,'title': self.title,'description': self.description,'status': self.status,'created_at': self.created_at}@classmethoddef from_dict(cls, data: Dict[str, Any]) -> 'Task':"""从字典创建任务"""return cls(id=data['id'],title=data['title'],description=data['description'],status=data['status'],created_at=data['created_at'])class TaskRepository(ABC):"""任务仓储接口 - 数据访问抽象"""@abstractmethoddef add(self, task: Task) -> None:pass@abstractmethoddef get_by_id(self, task_id: str) -> Optional[Task]:pass@abstractmethoddef get_all(self) -> List[Task]:pass@abstractmethoddef update(self, task: Task) -> None:pass@abstractmethoddef delete(self, task_id: str) -> bool:passclass TaskFilter:"""任务过滤器 - 值对象"""def __init__(self, status: str = "all"):self.status = status  # "all", "pending", "completed"def apply(self, tasks: List[Task]) -> List[Task]:"""应用过滤器"""if self.status == "all":return taskselif self.status == "pending":return [task for task in tasks if task.status == "pending"]elif self.status == "completed":return [task for task in tasks if task.status == "completed"]return tasks# ======================
# 应用层 (Application Layer)
# ======================class TaskService:"""任务服务 - 业务逻辑"""def __init__(self, task_repository: TaskRepository):self.task_repository = task_repositorydef create_task(self, title: str, description: str) -> Task:"""创建新任务"""if not title or not description:raise ValueError("标题和描述不能为空")task_id = str(uuid.uuid4())task = Task(id=task_id, title=title, description=description)self.task_repository.add(task)return taskdef get_task(self, task_id: str) -> Optional[Task]:"""获取任务"""return self.task_repository.get_by_id(task_id)def get_all_tasks(self, filter: TaskFilter = None) -> List[Task]:"""获取所有任务"""tasks = self.task_repository.get_all()if filter:tasks = filter.apply(tasks)return tasksdef toggle_task_status(self, task_id: str) -> Optional[Task]:"""切换任务状态"""task = self.task_repository.get_by_id(task_id)if task:task.toggle_status()self.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, int]:"""获取任务统计"""tasks = self.task_repository.get_all()total = len(tasks)completed = len([task for task in tasks if task.is_completed()])pending = total - completedreturn {'total': total,'completed': completed,'pending': pending}# ======================
# 接口适配器层 (Interface Adapters)
# ======================class InMemoryTaskRepository(TaskRepository):"""内存任务仓储实现"""def __init__(self):self.tasks: Dict[str, Task] = {}def add(self, task: Task) -> None:self.tasks[task.id] = taskdef get_by_id(self, task_id: str) -> Optional[Task]:return self.tasks.get(task_id)def get_all(self) -> List[Task]:return list(self.tasks.values())def update(self, task: Task) -> 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 Falseclass TaskPresenter:"""任务展示器 - 准备视图所需的数据"""@staticmethoddef present_task(task: Task) -> Dict[str, Any]:"""展示单个任务"""return task.to_dict()@staticmethoddef present_task_list(tasks: List[Task]) -> List[Dict[str, Any]]:"""展示任务列表"""return [task.to_dict() for task in tasks]@staticmethoddef present_statistics(statistics: Dict[str, int]) -> Dict[str, Any]:"""展示统计信息"""return {'total_tasks': statistics['total'],'completed_tasks': statistics['completed'],'pending_tasks': statistics['pending'],'completion_rate': (round(statistics['completed'] / statistics['total'] * 100, 2) if statistics['total'] > 0 else 0)}class TaskController:"""任务控制器 - 处理用户输入"""def __init__(self, task_service: TaskService):self.task_service = task_serviceself.presenter = TaskPresenter()def create_task(self, title: str, description: str) -> Dict[str, Any]:"""创建任务"""try:task = self.task_service.create_task(title, description)return {'success': True,'task': self.presenter.present_task(task),'message': '任务创建成功'}except ValueError as e:return {'success': False,'task': None,'message': str(e)}def get_tasks(self, filter_status: str = "all") -> Dict[str, Any]:"""获取任务列表"""task_filter = TaskFilter(status=filter_status)tasks = self.task_service.get_all_tasks(task_filter)return {'success': True,'tasks': self.presenter.present_task_list(tasks),'filter': filter_status}def toggle_task_status(self, task_id: str) -> Dict[str, Any]:"""切换任务状态"""task = self.task_service.toggle_task_status(task_id)if task:return {'success': True,'task': self.presenter.present_task(task),'message': '任务状态已更新'}else:return {'success': False,'task': None,'message': '任务未找到'}def delete_task(self, task_id: str) -> Dict[str, Any]:"""删除任务"""success = self.task_service.delete_task(task_id)if success:return {'success': True,'message': '任务已删除'}else:return {'success': False,'message': '任务未找到'}def get_statistics(self) -> Dict[str, Any]:"""获取统计信息"""statistics = self.task_service.get_task_statistics()return {'success': True,'statistics': self.presenter.present_statistics(statistics)}# ======================
# 框架层 (Frameworks & Drivers)
# ======================class TaskManagerApp:"""Dash 应用 - UI 框架层"""def __init__(self, controller: TaskController):self.controller = controllerself.app = dash.Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])self.setup_layout()self.setup_callbacks()def setup_layout(self):"""设置应用布局"""self.app.layout = dbc.Container([# 标题区域dbc.Row([dbc.Col([html.H1("任务管理系统 - 整洁架构示例", className="text-center my-4 text-primary")])]),# 统计信息区域dbc.Row([dbc.Col([dbc.Card([dbc.CardBody([dbc.Row([dbc.Col([html.H5("总任务数", className="card-title"),html.H3(id="total-tasks", className="card-text text-primary")], width=3),dbc.Col([html.H5("待完成", className="card-title"),html.H3(id="pending-tasks", className="card-text text-warning")], width=3),dbc.Col([html.H5("已完成", className="card-title"),html.H3(id="completed-tasks", className="card-text text-success")], width=3),dbc.Col([html.H5("完成率", className="card-title"),html.H3(id="completion-rate", className="card-text text-info")], width=3),])])], 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"),dbc.Input(id="task-desc-input",placeholder="任务描述",type="text",className="mb-2"),], width=10),dbc.Col([dbc.Button("添加任务", id="add-task-btn", color="primary",className="w-100 h-100")], width=2)]),html.Div(id="add-task-message")])], 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),]),], width=8),dbc.Col([dbc.Button("刷新统计", id="refresh-stats", color="info", className="float-end")], width=4)])])], className="mb-4")])]),# 任务列表区域dbc.Row([dbc.Col([html.Div(id="tasks-container")])]),# 存储组件dcc.Store(id="tasks-store", data=[]),dcc.Store(id="filter-store", data="all")], fluid=True)def setup_callbacks(self):"""设置应用回调"""# 初始化统计信息和任务列表@self.app.callback([Output("tasks-store", "data"),Output("total-tasks", "children"),Output("pending-tasks", "children"),Output("completed-tasks", "children"),Output("completion-rate", "children")],[Input("filter-all", "n_clicks"),Input("filter-pending", "n_clicks"),Input("filter-completed", "n_clicks"),Input("refresh-stats", "n_clicks")])def update_tasks_and_stats(all_clicks, pending_clicks, completed_clicks, refresh_clicks):"""更新任务列表和统计信息"""ctx = callback_contextfilter_status = "all"if ctx.triggered:trigger_id = ctx.triggered[0]['prop_id'].split('.')[0]if trigger_id == "filter-pending":filter_status = "pending"elif trigger_id == "filter-completed":filter_status = "completed"# 获取任务列表tasks_result = self.controller.get_tasks(filter_status)tasks_data = tasks_result['tasks'] if tasks_result['success'] else []# 获取统计信息stats_result = self.controller.get_statistics()if stats_result['success']:stats = stats_result['statistics']total = stats['total_tasks']pending = stats['pending_tasks']completed = stats['completed_tasks']completion_rate = f"{stats['completion_rate']}%"else:total = pending = completed = 0completion_rate = "0%"return tasks_data, total, pending, completed, completion_rate# 添加任务@self.app.callback([Output("add-task-message", "children"),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")])def add_task(n_clicks, title, description):"""添加新任务"""if n_clicks is None or n_clicks == 0:return "", "", ""if not title or not description:return dbc.Alert("请输入任务标题和描述", color="danger"), title, descriptionresult = self.controller.create_task(title, description)if result['success']:return dbc.Alert(result['message'], color="success"), "", ""else:return dbc.Alert(result['message'], color="danger"), title, description# 渲染任务列表@self.app.callback(Output("tasks-container", "children"),[Input("tasks-store", "data")])def render_tasks(tasks_data):"""渲染任务列表"""if not tasks_data:return dbc.Alert("暂无任务", color="info")task_cards = []for task in tasks_data:status_color = "success" if task['status'] == 'completed' else "warning"status_text = "已完成" if task['status'] == 'completed' else "待完成"card = dbc.Card([dbc.CardBody([dbc.Row([dbc.Col([html.H5(task['title']),html.P(task['description'], className="text-muted"),html.Small(f"创建时间: {task['created_at']}", className="text-muted")], width=8),dbc.Col([dbc.Badge(status_text, color=status_color, className="mb-2 d-block"),html.Div([dbc.Button("切换状态", id={"type": "toggle-btn", "index": task['id']},color="primary",size="sm",className="me-1 mb-1"),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("tasks-store", "data", allow_duplicate=True),Output("total-tasks", "children", allow_duplicate=True),Output("pending-tasks", "children", allow_duplicate=True),Output("completed-tasks", "children", allow_duplicate=True),Output("completion-rate", "children", allow_duplicate=True)],[Input({"type": "toggle-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, delete_clicks):"""处理任务操作"""ctx = callback_contextif not ctx.triggered:return dash.no_update, dash.no_update, dash.no_update, dash.no_update, dash.no_updatetrigger_id = ctx.triggered[0]['prop_id']# 解析触发按钮的类型和IDif 'toggle-btn' in trigger_id:task_id = eval(trigger_id.split('.')[0])['index']self.controller.toggle_task_status(task_id)elif 'delete-btn' in trigger_id:task_id = eval(trigger_id.split('.')[0])['index']self.controller.delete_task(task_id)# 更新任务列表和统计信息tasks_result = self.controller.get_tasks()tasks_data = tasks_result['tasks'] if tasks_result['success'] else []stats_result = self.controller.get_statistics()if stats_result['success']:stats = stats_result['statistics']total = stats['total_tasks']pending = stats['pending_tasks']completed = stats['completed_tasks']completion_rate = f"{stats['completion_rate']}%"else:total = pending = completed = 0completion_rate = "0%"return tasks_data, total, pending, completed, completion_ratedef run(self, debug=True):"""运行应用"""self.app.run(debug=debug)# ======================
# 依赖注入和应用启动
# ======================def create_app():"""创建应用实例 - 依赖注入容器"""# 创建仓储task_repository = InMemoryTaskRepository()# 创建服务task_service = TaskService(task_repository)# 创建控制器task_controller = TaskController(task_service)# 创建Dash应用app = TaskManagerApp(task_controller)return appif __name__ == "__main__":# 创建并运行应用app = create_app()app.run(debug=True)