MVP 架构模式

一、MVP 模式概述

MVP (Model-View-Presenter) 是一种软件架构模式,用于分离用户界面、业务逻辑和数据管理。

核心思想

• 关注点分离: 将应用分为三个独立的部分
• 可测试性: 各部分可独立测试
• 可维护性: 职责清晰,易于修改和扩展

二、三个角色及职责

Model (模型)

职责: 管理数据和业务逻辑

包含:

• 数据的增删改查
• 业务规则和验证
• 数据持久化 (数据库、文件、网络)
• 状态管理
• 数据计算和处理

不包含:

• ❌ 任何 UI 相关代码
• ❌ 不知道 View 和 Presenter 的存在
• ❌ 不决定如何显示数据

View (视图)

职责: 显示界面和接收用户输入

包含:

• UI 控件的创建和布局
• 接收用户操作 (点击、输入等)
• 显示数据 (被动接收命令)
• UI 状态切换 (显示/隐藏、启用/禁用)
• 动画和样式

不包含:

• ❌ 业务逻辑
• ❌ 数据处理
• ❌ 不直接访问 Model

Presenter (展示器)

职责: 协调 View 和 Model,处理交互逻辑

包含:

• 响应 View 的用户操作
• 调用 Model 的业务逻辑
• 处理数据格式化和转换
• 决定显示什么内容
• 控制 UI 流程

不包含:

• ❌ 不创建 UI 控件
• ❌ 不直接操作数据库
• ❌ 不包含核心业务逻辑 (委托给 Model)

三、数据流向

用户操作 → View → Presenter → Model
                    ↓
         View ← Presenter ← Model (通知数据变化)

关键点:

• View 和 Model 互不直接通信
• 所有交互通过 Presenter 中转
• Model 通过信号/事件通知变化,Presenter 监听并更新 View

四、完整示例: 待办事项应用

1. Model 层实现

from PySide6.QtCore import QObject, Signal
import sqlite3
from datetime import datetime
 
class Task:
    """数据实体"""
    def __init__(self, id, title, completed=False, created_at=None):
        self.id = id
        self.title = title
        self.completed = completed
        self.created_at = created_at or datetime.now()
 
class TaskModel(QObject):
    """任务数据模型"""
    # 定义信号:通知数据变化
    tasks_loaded = Signal(list)
    task_added = Signal(object)
    task_updated = Signal(object)
    task_deleted = Signal(int)
    error_occurred = Signal(str)
    
    def __init__(self, db_path='tasks.db'):
        super().__init__()
        self.db_path = db_path
        self._tasks = []
        self._init_database()
    
    # === 数据持久化 ===
    def _init_database(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                completed BOOLEAN DEFAULT 0,
                created_at TIMESTAMP
            )
        ''')
        conn.commit()
        conn.close()
    
    # === 数据查询 ===
    def load_all_tasks(self):
        """加载所有任务"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM tasks ORDER BY created_at DESC')
            rows = cursor.fetchall()
            conn.close()
            
            self._tasks = [
                Task(id=r[0], title=r[1], completed=bool(r[2]), 
                     created_at=r[3])
                for r in rows
            ]
            self.tasks_loaded.emit(self._tasks)
        except Exception as e:
            self.error_occurred.emit(f"加载任务失败: {e}")
    
    def get_task(self, task_id):
        """获取单个任务"""
        return next((t for t in self._tasks if t.id == task_id), None)
    
    # === 业务逻辑 ===
    def add_task(self, title):
        """添加任务"""
        # 验证逻辑
        if not title or len(title.strip()) == 0:
            self.error_occurred.emit("任务标题不能为空")
            return False
        
        if len(title) > 100:
            self.error_occurred.emit("任务标题不能超过100字符")
            return False
        
        # 保存到数据库
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            cursor.execute(
                'INSERT INTO tasks (title, completed, created_at) VALUES (?, ?, ?)',
                (title.strip(), False, datetime.now())
            )
            conn.commit()
            task_id = cursor.lastrowid
            conn.close()
            
            # 创建任务对象
            task = Task(task_id, title.strip(), False)
            self._tasks.insert(0, task)  # 新任务在最前面
            
            # 通知添加成功
            self.task_added.emit(task)
            return True
        except Exception as e:
            self.error_occurred.emit(f"添加任务失败: {e}")
            return False
    
    def toggle_task(self, task_id):
        """切换任务完成状态"""
        task = self.get_task(task_id)
        if not task:
            self.error_occurred.emit("任务不存在")
            return
        
        try:
            # 更新状态
            task.completed = not task.completed
            
            # 保存到数据库
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            cursor.execute(
                'UPDATE tasks SET completed = ? WHERE id = ?',
                (task.completed, task_id)
            )
            conn.commit()
            conn.close()
            
            # 通知更新
            self.task_updated.emit(task)
        except Exception as e:
            self.error_occurred.emit(f"更新任务失败: {e}")
    
    def delete_task(self, task_id):
        """删除任务"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            cursor.execute('DELETE FROM tasks WHERE id = ?', (task_id,))
            conn.commit()
            conn.close()
            
            # 从内存中删除
            self._tasks = [t for t in self._tasks if t.id != task_id]
            
            # 通知删除
            self.task_deleted.emit(task_id)
        except Exception as e:
            self.error_occurred.emit(f"删除任务失败: {e}")
    
    # === 业务查询 ===
    def get_pending_count(self):
        """获取未完成任务数量"""
        return sum(1 for t in self._tasks if not t.completed)
    
    def get_completed_count(self):
        """获取已完成任务数量"""
        return sum(1 for t in self._tasks if t.completed)

2. View 层实现

from PySide6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, 
                               QLineEdit, QPushButton, QListWidget, 
                               QListWidgetItem, QCheckBox, QLabel,
                               QMessageBox)
from PySide6.QtCore import Signal
 
class TaskView(QWidget):
    """任务视图:只负责显示和接收用户操作"""
    
    # 定义信号:通知 Presenter 用户操作
    add_task_requested = Signal(str)
    task_toggled = Signal(int)
    task_deleted = Signal(int)
    load_tasks_requested = Signal()
    
    def __init__(self):
        super().__init__()
        self._init_ui()
    
    def _init_ui(self):
        """初始化 UI 控件"""
        self.setWindowTitle("待办事项")
        self.setMinimumSize(500, 600)
        
        layout = QVBoxLayout()
        
        # 标题
        title_label = QLabel("我的待办事项")
        title_label.setStyleSheet("font-size: 24px; font-weight: bold;")
        layout.addWidget(title_label)
        
        # 统计信息
        self.stats_label = QLabel("未完成: 0 | 已完成: 0")
        layout.addWidget(self.stats_label)
        
        # 输入框和添加按钮
        input_layout = QHBoxLayout()
        self.task_input = QLineEdit()
        self.task_input.setPlaceholderText("输入新任务...")
        self.task_input.returnPressed.connect(self._on_add_clicked)
        
        self.add_button = QPushButton("添加")
        self.add_button.clicked.connect(self._on_add_clicked)
        
        input_layout.addWidget(self.task_input)
        input_layout.addWidget(self.add_button)
        layout.addLayout(input_layout)
        
        # 任务列表
        self.task_list = QListWidget()
        layout.addWidget(self.task_list)
        
        self.setLayout(layout)
        
        # 发送加载请求
        self.load_tasks_requested.emit()
    
    # === 处理用户操作:发送信号 ===
    def _on_add_clicked(self):
        """用户点击添加按钮"""
        title = self.task_input.text().strip()
        if title:
            self.add_task_requested.emit(title)  # 通知 Presenter
    
    def _on_task_toggled(self, task_id):
        """用户切换任务状态"""
        self.task_toggled.emit(task_id)
    
    def _on_task_deleted(self, task_id):
        """用户删除任务"""
        self.task_deleted.emit(task_id)
    
    # === 接收 Presenter 的命令:更新显示 ===
    def display_tasks(self, tasks):
        """显示任务列表"""
        self.task_list.clear()
        
        for task in tasks:
            item = QListWidgetItem(self.task_list)
            item.setData(1, task.id)  # 存储任务 ID
            
            # 创建自定义 widget
            item_widget = self._create_task_widget(task)
            item.setSizeHint(item_widget.sizeHint())
            
            self.task_list.addItem(item)
            self.task_list.setItemWidget(item, item_widget)
    
    def _create_task_widget(self, task):
        """创建任务显示控件"""
        widget = QWidget()
        layout = QHBoxLayout()
        layout.setContentsMargins(5, 5, 5, 5)
        
        # 复选框
        checkbox = QCheckBox()
        checkbox.setChecked(task.completed)
        checkbox.stateChanged.connect(
            lambda: self._on_task_toggled(task.id)
        )
        
        # 任务标题
        title_label = QLabel(task.title)
        if task.completed:
            title_label.setStyleSheet("text-decoration: line-through; color: gray;")
        
        # 删除按钮
        delete_button = QPushButton("删除")
        delete_button.clicked.connect(
            lambda: self._on_task_deleted(task.id)
        )
        
        layout.addWidget(checkbox)
        layout.addWidget(title_label, stretch=1)
        layout.addWidget(delete_button)
        
        widget.setLayout(layout)
        return widget
    
    def clear_input(self):
        """清空输入框"""
        self.task_input.clear()
        self.task_input.setFocus()
    
    def update_statistics(self, pending_count, completed_count):
        """更新统计信息"""
        self.stats_label.setText(
            f"未完成: {pending_count} | 已完成: {completed_count}"
        )
    
    def show_error(self, message):
        """显示错误消息"""
        QMessageBox.warning(self, "错误", message)
    
    def show_success(self, message):
        """显示成功消息"""
        QMessageBox.information(self, "成功", message)

3. Presenter 层实现

class TaskPresenter:
    """任务展示器:协调 View 和 Model"""
    
    def __init__(self, view, model):
        self.view = view
        self.model = model
        self._connect_signals()
    
    def _connect_signals(self):
        """连接信号和槽"""
        # View 的信号 → Presenter 的方法
        self.view.load_tasks_requested.connect(self.load_tasks)
        self.view.add_task_requested.connect(self.add_task)
        self.view.task_toggled.connect(self.toggle_task)
        self.view.task_deleted.connect(self.delete_task)
        
        # Model 的信号 → Presenter 的方法
        self.model.tasks_loaded.connect(self.on_tasks_loaded)
        self.model.task_added.connect(self.on_task_added)
        self.model.task_updated.connect(self.on_task_updated)
        self.model.task_deleted.connect(self.on_task_deleted)
        self.model.error_occurred.connect(self.on_error)
    
    # === 响应 View 的操作 ===
    def load_tasks(self):
        """加载任务"""
        self.model.load_all_tasks()
    
    def add_task(self, title):
        """添加任务"""
        # Presenter 可以添加额外的逻辑
        if self.model.add_task(title):
            # 添加成功后的 UI 操作
            self.view.clear_input()
    
    def toggle_task(self, task_id):
        """切换任务状态"""
        self.model.toggle_task(task_id)
    
    def delete_task(self, task_id):
        """删除任务"""
        # Presenter 可以添加确认逻辑
        from PySide6.QtWidgets import QMessageBox
        reply = QMessageBox.question(
            self.view, 
            "确认删除", 
            "确定要删除这个任务吗?",
            QMessageBox.Yes | QMessageBox.No
        )
        
        if reply == QMessageBox.Yes:
            self.model.delete_task(task_id)
    
    # === 响应 Model 的变化 ===
    def on_tasks_loaded(self, tasks):
        """任务加载完成"""
        # 数据格式化和转换(如果需要)
        self.view.display_tasks(tasks)
        self._update_statistics()
    
    def on_task_added(self, task):
        """任务添加成功"""
        # 重新加载列表
        self.model.load_all_tasks()
        # 可以显示成功提示
        # self.view.show_success("任务添加成功")
    
    def on_task_updated(self, task):
        """任务更新成功"""
        self.model.load_all_tasks()
        self._update_statistics()
    
    def on_task_deleted(self, task_id):
        """任务删除成功"""
        self.model.load_all_tasks()
        self._update_statistics()
    
    def on_error(self, message):
        """处理错误"""
        self.view.show_error(message)
    
    # === 辅助方法 ===
    def _update_statistics(self):
        """更新统计信息"""
        pending = self.model.get_pending_count()
        completed = self.model.get_completed_count()
        self.view.update_statistics(pending, completed)

4. 主程序入口

import sys
from PySide6.QtWidgets import QApplication
 
def main():
    app = QApplication(sys.argv)
    
    # 创建三个组件
    model = TaskModel()
    view = TaskView()
    presenter = TaskPresenter(view, model)
    
    # 显示视图
    view.show()
    
    sys.exit(app.exec())
 
if __name__ == '__main__':
    main()

五、MVP 模式的优势

1. 可测试性

# 可以单独测试 Model
def test_add_task():
    model = TaskModel(':memory:')  # 使用内存数据库
    model.add_task("测试任务")
    assert model.get_pending_count() == 1
 
# 可以 Mock View 测试 Presenter
class MockView:
    def __init__(self):
        self.displayed_tasks = []
    
    def display_tasks(self, tasks):
        self.displayed_tasks = tasks
 
def test_presenter():
    view = MockView()
    model = TaskModel(':memory:')
    presenter = TaskPresenter(view, model)
    
    model.add_task("任务1")
    assert len(view.displayed_tasks) == 1

2. 职责清晰

• Model: 只关心数据和业务
• View: 只关心显示
• Presenter: 协调两者

3. 易于维护

• 修改 UI 不影响业务逻辑
• 修改业务逻辑不影响 UI
• 可以替换实现 (如换数据库、换 UI 框架)

4. 代码复用

• Model 可以在多个界面中复用
• 同一个 Model 可以对应多个 View

六、最佳实践

1. View 应该 ” 哑 “

# ❌ 错误:View 包含业务逻辑
class BadView(QWidget):
    def on_button_clicked(self):
        if self.input.text().strip() == "":
            QMessageBox.warning(self, "错误", "不能为空")
            return
        # 处理数据...
 
# ✅ 正确:View 只发信号
class GoodView(QWidget):
    button_clicked = Signal(str)
    
    def on_button_clicked(self):
        self.button_clicked.emit(self.input.text())

2. Model 应该独立

# ✅ Model 可以脱离 GUI 独立运行
if __name__ == '__main__':
    model = TaskModel()
    model.add_task("命令行任务")
    tasks = model._tasks
    for task in tasks:
        print(f"- [{' x' if task.completed else '  '}] {task.title}")

3. Presenter 是胶水代码

• 不要在 Presenter 中编写复杂的业务逻辑 (应该在 Model)
• 不要在 Presenter 中创建 UI 控件 (应该在 View)
• Presenter 主要做: 转发、协调、格式化

4. 使用信号解耦

# Model 不知道谁在监听
class Model(QObject):
    data_changed = Signal()
    
    def update(self):
        # ...
        self.data_changed.emit()  # 只管发信号
 
# Presenter 监听并处理
class Presenter:
    def __init__(self, view, model):
        model.data_changed.connect(self.on_data_changed)

七、常见问题

Q1: Presenter 和 Model 都有业务逻辑,怎么区分?

• Model: 核心业务规则 (如计算、验证、数据操作)
• Presenter:UI 相关的业务流程 (如 ” 先验证,再保存,再刷新界面 “)

Q2: View 能否直接绑定 Model?

• Passive View: 不可以,完全隔离
• Supervising Controller: 可以简单绑定,复杂逻辑走 Presenter

Q3: 一个 View 可以有多个 Presenter 吗?

• 通常一个 View 对应一个 Presenter
• 如果功能复杂,可以将 View 拆分成多个子 View

Q4: Model 之间可以相互调用吗?

• 可以,但建议通过 Service 层或 Facade 模式协调
• 避免 Model 之间强耦合

八、总结

MVP 三要素

1. Model: 我有什么数据 (What)
2. View: 怎么显示 (How to display)
3. Presenter: 显示什么,什么时候显示 (What to display, When)

核心原则

• View 和 Model 互不直接通信
• Presenter 是中间协调者
• 通过信号/事件解耦

适用场景

• 桌面 GUI 应用 (Qt, WPF, WinForms)
• 移动应用 (Android)
• 需要高可测试性的项目
• 团队协作 (UI、业务逻辑可并行开发)