从零搭建Python个人待办事项系统：轻量级架构与本地部署

一、选型考量与认知辨析

在设计个人待办系统时，开发者往往容易走向两个极端：要么简单地使用单个 TXT 或 JSON 文件保存数据，要么直接引入 MySQL/PostgreSQL 配合 Docker 容器。这两种方案看似合理，但在个人单机应用场景下都存在明显的结构性缺陷。

1. 为什么“直接读写 JSON 文件”在待办场景中不够稳妥？

纯文本或 JSON 文件的优势在于直观透明，任何文本编辑器都能直接打开查看，对于初学者几乎没有学习成本。在只有几条静态配置读取的场景下，这确实是最简单的做法。

但在待办事项系统中，数据的典型行为是高频碎片化修改（例如迅速打勾完成一项、微调优先级、频繁增删）。基于普通文件读写时，每一次修改都必须经历“全量读取文件 → 反序列化为 Python 字典列表 → 在内存中增删改 → 序列化为 JSON 字符串 → 覆盖写入磁盘文件”。

这种模式存在三个关键隐患：

• 断电或异常退出的写入截断：如果操作系统在写盘的瞬间断电或进程被杀，原有文件将被写入一半或截断为 0 字节，造成全量历史数据丢失。
• 缺乏原子性事务保护：无法在底层锁机制下提供强一致性保证，并发读写容易出现脏数据覆盖。
• 低效的全量重写：随着完成的待办事项累积至数千条，仅仅勾选一个任务就需要重新写出几十万字符的文本，产生不必要的 I/O 磨损与卡顿。

2. 为什么在个人系统中没必要引入独立数据库服务？

另一派做法是套用企业级开发惯性，安装 MySQL、PostgreSQL 或通过 Docker 启动 Redis 和服务编排。对于多用户、多终端并发写入、团队协作的商业系统，独立数据库服务提供了精细的权限管理与网络并发能力，这无疑是正确的。

但在本地个人系统里，这种做法属于典型的过度设计：独立数据库需要常驻后台进程，白白吞噬 100MB 至 300MB 的系统内存；需要管理数据库账号、密码和连接池；一旦本机环境变动或系统更新，经常遇到端口冲突或服务启动超时问题；更重要的是，单机迁移与备份必须借助专用 dump 工具导出 SQL，无法直接随手复制。

3. 本地嵌入式 SQLite 的黄金平衡点

SQLite 恰恰吸收了上述两者的优势并规避了劣势：它随 Python 解释器原生内置，不需要安装任何独立软件；数据库直接体现为一个普通的单文件（如 todos.db），备份只需直接复制；其内部由高度成熟的 B 树存储引擎驱动，提供完整的 ACID 事务安全，写入具有回滚日志保护；在单机数万条待办记录下，查询与更新耗时均在 0.5 毫秒以内。

二、系统分层架构与数据模型设计

本项目坚持“零外部安装包”与“三层极简解耦”的设计原则。系统划分为客户端展现层、核心服务路由层与数据持久层，层与层之间边界清晰，单一职责明确。

1. 架构各层职责划分

• 客户端展现层（UI）：采用无框架依赖的单一 index.html 文件，内嵌 CSS3 样式与原生 Vanilla JavaScript。通过浏览器的 fetch() API 与后端异步通信，实现任务的即时创建、切换完成状态、按标签筛选与局部 DOM 刷新，避免整页重新加载。
• 核心服务路由层（Server）：利用 Python 标准库 http.server.ThreadingHTTPServer，侦听 127.0.0.1:8080。其职责涵盖两部分：静态文件挂载分发，以及 /api/todos 系列 RESTful 端点的路由分发、JSON 参数解析和 HTTP 状态码响应。
• 数据持久层（DAO）：通过 Python 标准库 sqlite3 封装专属数据库操作类，统一接管数据表初始化、SQL 参数化过滤与连接上下文管理，彻底杜绝 SQL 注入隐患。

2. 核心数据表设计

我们在 SQLite 中建立一张名为 todos 的数据表，字段覆盖日常任务管理的核心属性：

三、核心实现：从零编写项目代码

按照自底向上的实际开发顺序，我们先建立规范的项目目录，接着编写数据持久层，再编写 HTTP 调度层，最后完成交互前端。

1. 规划项目目录结构

在任意本地工作目录下建立项目工程目录 todo_project。该系统仅由两个 Python 文件和一个前端单文件构成，结构极其规整：

todo_project/
├── database.py       # 数据持久层：封装 SQLite 操作与 CRUD 接口
├── server.py         # 服务调度层：RESTful API 路由与静态资源托管
├── static/
│   └── index.html    # 交互展现层：单页面待办操作前端
└── todos.db          # 运行时由程序自动创建的本地 SQLite 数据库文件

2. 第一步：封装数据库操作模块（database.py）

创建 database.py 文件。该模块通过 Python 的 sqlite3 模块实现对数据库的全部交互，包含自动建表、开启 WAL 并发优化以及完整的增删改查方法：

import sqlite3
import os
from typing import List, Dict, Optional, Any

DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "todos.db")

class Database:
    def __init__(self, db_path: str = DB_PATH):
        self.db_path = db_path
        self._init_db()

    def _get_connection(self) -> sqlite3.Connection:
        """获取数据库连接并配置行记录格式与 WAL 模式"""
        conn = sqlite3.connect(self.db_path)
        # 使查询结果可以通过列名访问（dict-like row）
        conn.row_factory = sqlite3.Row
        # 开启 WAL 模式提升并发读写性能与抗锁能力
        conn.execute("PRAGMA journal_mode=WAL;")
        return conn

    def _init_db(self):
        """初始化数据表结构"""
        with self._get_connection() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS todos (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT NOT NULL,
                    description TEXT DEFAULT '',
                    priority TEXT CHECK(priority IN ('low', 'medium', 'high')) DEFAULT 'medium',
                    status TEXT CHECK(status IN ('pending', 'completed')) DEFAULT 'pending',
                    created_at TEXT DEFAULT (datetime('now', 'localtime'))
                );
            """)
            conn.commit()

    def list_todos(self, status: Optional[str] = None, priority: Optional[str] = None) -> List[Dict[str, Any]]:
        """条件筛选查询待办列表"""
        query = "SELECT * FROM todos WHERE 1=1"
        params = []
        if status:
            query += " AND status = ?"
            params.append(status)
        if priority:
            query += " AND priority = ?"
            params.append(priority)
        
        # 默认按照待办优先、按优先级高低及创建时间倒序排布
        query += " ORDER BY CASE status WHEN 'pending' THEN 0 ELSE 1 END, "                  "CASE priority WHEN 'high' THEN 1 WHEN 'medium' THEN 2 ELSE 3 END, id DESC"

        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(query, params)
            return [dict(row) for row in cursor.fetchall()]

    def add_todo(self, title: str, description: str = "", priority: str = "medium") -> int:
        """创建新待办项，返回新生成的自增主键 ID"""
        clean_title = title.strip()
        if not clean_title:
            raise ValueError("待办事项标题不能为空")
        if priority not in ('low', 'medium', 'high'):
            priority = 'medium'

        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                "INSERT INTO todos (title, description, priority) VALUES (?, ?, ?);",
                (clean_title, description.strip(), priority)
            )
            conn.commit()
            return cursor.lastrowid

    def update_todo(self, todo_id: int, status: Optional[str] = None, title: Optional[str] = None, 
                    description: Optional[str] = None, priority: Optional[str] = None) -> bool:
        """根据 ID 局部更新待办事项属性"""
        updates = []
        params = []
        if status is not None and status in ('pending', 'completed'):
            updates.append("status = ?")
            params.append(status)
        if title is not None and title.strip():
            updates.append("title = ?")
            params.append(title.strip())
        if description is not None:
            updates.append("description = ?")
            params.append(description.strip())
        if priority is not None and priority in ('low', 'medium', 'high'):
            updates.append("priority = ?")
            params.append(priority)

        if not updates:
            return False

        params.append(todo_id)
        query = f"UPDATE todos SET {', '.join(updates)} WHERE id = ?;"

        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(query, params)
            conn.commit()
            return cursor.rowcount > 0

    def delete_todo(self, todo_id: int) -> bool:
        """根据 ID 永久物理删除一项待办"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM todos WHERE id = ?;", (todo_id,))
            conn.commit()
            return cursor.rowcount > 0

3. 第二步：编写 RESTful 服务端与路由分发（server.py）

创建 server.py 文件。该文件使用 Python 内置的 ThreadingHTTPServer，确保多个异步请求互不阻塞，并针对 /api/todos 端点分别映射对应的 HTTP 方法与状态码：

import json
import os
import re
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
from database import Database

HOST = "127.0.0.1"  # 严格锁定本机回环，杜绝无鉴权外部暴露
PORT = 8080
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
STATIC_DIR = os.path.join(BASE_DIR, "static")

db = Database()

class TodoRequestHandler(SimpleHTTPRequestHandler):
    def __init__(self, *args, **kwargs):
        # 指定静态资源托管物理目录
        super().__init__(*args, directory=STATIC_DIR, **kwargs)

    def _send_json_response(self, data: Any, status: int = HTTPStatus.OK):
        """格式化发送 JSON 响应头与数据体"""
        body = json.dumps(data, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-cache, no-store, must-revalidate")
        self.end_headers()
        self.wfile.write(body)

    def _parse_json_body(self) -> Dict[str, Any]:
        """安全读取并反序列化 HTTP 请求体中的 JSON 数据"""
        content_length = int(self.headers.get("Content-Length", 0))
        if content_length <= 0:
            return {}
        raw_data = self.rfile.read(content_length).decode("utf-8")
        return json.loads(raw_data)

    def do_GET(self):
        parsed_url = urlparse(self.path)
        path = parsed_url.path

        # 根路径重定向或返回 index.html
        if path == "/":
            self.path = "/index.html"
            return super().do_GET()

        # REST API: 获取待办事项列表
        if path == "/api/todos":
            params = parse_qs(parsed_url.query)
            status_filter = params.get("status", [None])[0]
            priority_filter = params.get("priority", [None])[0]
            todos = db.list_todos(status=status_filter, priority=priority_filter)
            self._send_json_response({"success": True, "data": todos})
            return

        # 托管其他静态静态资源
        return super().do_GET()

    def do_POST(self):
        parsed_url = urlparse(self.path)
        if parsed_url.path == "/api/todos":
            try:
                payload = self._parse_json_body()
                title = payload.get("title", "")
                desc = payload.get("description", "")
                priority = payload.get("priority", "medium")

                new_id = db.add_todo(title=title, description=desc, priority=priority)
                self._send_json_response(
                    {"success": True, "message": "创建成功", "id": new_id},
                    status=HTTPStatus.CREATED
                )
            except ValueError as ve:
                self._send_json_response({"success": False, "error": str(ve)}, status=HTTPStatus.BAD_REQUEST)
            except Exception as e:
                self._send_json_response({"success": False, "error": "服务端内部错误"}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
            return

        self._send_json_response({"success": False, "error": "端点不存在"}, status=HTTPStatus.NOT_FOUND)

    def do_PUT(self):
        # 匹配 /api/todos/<id> 格式
        match = re.match(r"^/api/todos/(\d+)$", self.path)
        if match:
            todo_id = int(match.group(1))
            try:
                payload = self._parse_json_body()
                ok = db.update_todo(
                    todo_id=todo_id,
                    status=payload.get("status"),
                    title=payload.get("title"),
                    description=payload.get("description"),
                    priority=payload.get("priority")
                )
                if ok:
                    self._send_json_response({"success": True, "message": "更新成功"})
                else:
                    self._send_json_response({"success": False, "error": "待办记录不存在或无变动"}, status=HTTPStatus.NOT_FOUND)
            except Exception as e:
                self._send_json_response({"success": False, "error": "更新处理失败"}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
            return

        self._send_json_response({"success": False, "error": "端点不存在"}, status=HTTPStatus.NOT_FOUND)

    def do_DELETE(self):
        # 匹配 /api/todos/<id> 格式
        match = re.match(r"^/api/todos/(\d+)$", self.path)
        if match:
            todo_id = int(match.group(1))
            ok = db.delete_todo(todo_id)
            if ok:
                self._send_json_response({"success": True, "message": "删除成功"})
            else:
                self._send_json_response({"success": False, "error": "待办记录不存在"}, status=HTTPStatus.NOT_FOUND)
            return

        self._send_json_response({"success": False, "error": "端点不存在"}, status=HTTPStatus.NOT_FOUND)

def run():
    # 确保静态资源目录存在
    os.makedirs(STATIC_DIR, exist_ok=True)
    server_address = (HOST, PORT)
    httpd = ThreadingHTTPServer(server_address, TodoRequestHandler)
    print(f"[*] 个人待办事项系统已启动: http://{HOST}:{PORT}")
    print(f"[*] 数据存储文件路径: {DB_PATH}")
    print("[*] 按 Ctrl + C 终止服务")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n[!] 接收到退出信号，服务正常关闭。")
        httpd.server_close()

if __name__ == "__main__":
    run()

4. 第三步：编写单页面交互前端（static/index.html）

在 static/ 目录下创建 index.html。页面采用原生 HTML5 + CSS3 + Fetch API，无需运行 Node.js 或打包编译工具，直接实现即时新增、状态勾选、删除与过滤：

<!DOCTYPE html>
<html lang="zh-CN">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>个人轻量待办事项系统</title>
  <style>
    :root { --bg: #F8F6F0; --card: #FFFFFF; --text: #3E423F; --border: #DFD9CE; --accent: #8A9478; }
    body { font-family: -apple-system, sans-serif; background: var(--bg); color: var(--text); padding: 24px; margin: 0; }
    .container { max-width: 680px; margin: 0 auto; background: var(--card); border: 1px solid var(--border); border-radius: 8px; padding: 24px; }
    h1 { font-size: 20px; margin-top: 0; border-bottom: 2px solid var(--accent); padding-bottom: 8px; }
    .form-group { display: flex; gap: 8px; margin-bottom: 16px; flex-wrap: wrap; }
    input[type="text"], select { padding: 8px 12px; border: 1px solid var(--border); border-radius: 4px; font-size: 14px; }
    input[type="text"] { flex: 1; min-width: 220px; }
    button { background: var(--accent); color: #FFF; border: none; padding: 8px 16px; border-radius: 4px; cursor: pointer; font-weight: bold; }
    button:hover { opacity: 0.9; }
    .filter-bar { display: flex; gap: 12px; margin-bottom: 16px; font-size: 13px; color: #666; align-items: center; }
    .filter-btn { background: none; border: 1px solid var(--border); color: #666; padding: 4px 10px; cursor: pointer; border-radius: 3px; }
    .filter-btn.active { background: var(--accent); color: #FFF; border-color: var(--accent); }
    .todo-list { list-style: none; padding: 0; margin: 0; }
    .todo-item { display: flex; align-items: center; justify-content: space-between; padding: 10px; border-bottom: 1px solid var(--border); }
    .todo-item.completed .title { text-decoration: line-through; color: #9E9E9E; }
    .todo-left { display: flex; align-items: center; gap: 10px; }
    .badge { font-size: 11px; padding: 2px 6px; border-radius: 3px; font-weight: bold; }
    .badge-high { background: #F8D7DA; color: #721C24; }
    .badge-medium { background: #FFF3CD; color: #856404; }
    .badge-low { background: #D1E7DD; color: #0F5132; }
    .del-btn { background: #E68585; padding: 4px 8px; font-size: 12px; }
  </style>
</head>
<body>
<div class="container">
  <h1>个人轻量待办事项系统</h1>
  <div class="form-group">
    <input type="text" id="newTitle" placeholder="输入待办事项概要..." />
    <select id="newPriority">
      <option value="medium">中优先级</option>
      <option value="high">高优先级</option>
      <option value="low">低优先级</option>
    </select>
    <button onclick="addTodo()">添加任务</button>
  </div>

  <div class="filter-bar">
    <span>状态筛选：</span>
    <button class="filter-btn active" onclick="setFilter('')">全部</button>
    <button class="filter-btn" onclick="setFilter('pending')">待办中</button>
    <button class="filter-btn" onclick="setFilter('completed')">已完成</button>
  </div>

  <ul id="todoList" class="todo-list"></ul>
</div>

<script>
  let currentFilter = '';

  async function fetchTodos() {
    const url = currentFilter ? `/api/todos?status=${currentFilter}` : '/api/todos';
    const res = await fetch(url);
    const result = await res.json();
    const list = document.getElementById('todoList');
    list.innerHTML = '';
    if (result.data.length === 0) {
      list.innerHTML = '<li style="padding: 20px; text-align: center; color: #999;">暂无对应事项</li>';
      return;
    }
    result.data.forEach(item => {
      const li = document.createElement('li');
      li.className = `todo-item ${item.status === 'completed' ? 'completed' : ''}`;
      li.innerHTML = `
        <div class="todo-left">
          <input type="checkbox" ${item.status === 'completed' ? 'checked' : ''} onchange="toggleStatus(${item.id}, this.checked)">
          <span class="badge badge-${item.priority}">${item.priority.toUpperCase()}</span>
          <span class="title">${escapeHtml(item.title)}</span>
        </div>
        <button class="del-btn" onclick="deleteTodo(${item.id})">删除</button>
      `;
      list.appendChild(li);
    });
  }

  async function addTodo() {
    const title = document.getElementById('newTitle').value.trim();
    const priority = document.getElementById('newPriority').value;
    if (!title) return alert('标题不能为空');
    await fetch('/api/todos', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ title, priority })
    });
    document.getElementById('newTitle').value = '';
    fetchTodos();
  }

  async function toggleStatus(id, isCompleted) {
    await fetch(`/api/todos/${id}`, {
      method: 'PUT',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ status: isCompleted ? 'completed' : 'pending' })
    });
    fetchTodos();
  }

  async function deleteTodo(id) {
    if (!confirm('确定彻底删除该项待办吗？')) return;
    await fetch(`/api/todos/${id}`, { method: 'DELETE' });
    fetchTodos();
  }

  function setFilter(status) {
    currentFilter = status;
    document.querySelectorAll('.filter-btn').forEach(btn => btn.classList.remove('active'));
    event.target.classList.add('active');
    fetchTodos();
  }

  function escapeHtml(str) {
    return str.replace(/[&<>"']/g, m => ({ '&': '&amp;', '<': '&lt;', '>': '&gt;', '"': '&quot;', "'": '&#39;' }[m]));
  }

  fetchTodos();
</script>
</body>
</html>

四、本地部署与持久化运行

完成代码编写后，需要保证服务能够稳定开机自启、后台静默常驻，并在发生异常或系统重启时自动恢复。

1. 交互验证与启动测试

打开本地终端，进入工程所在目录，直接执行启动命令：

python3 server.py

预期输出示例：

[*] 个人待办事项系统已启动: http://127.0.0.1:8080
[*] 数据存储文件路径: /Users/username/todo_project/todos.db
[*] 按 Ctrl + C 终止服务

此时在浏览器中访问 http://127.0.0.1:8080，即可看到完整的交互界面。尝试添加一条“阅读 Python 技术文档”并将其状态标记为已完成，随后关闭终端并重新打开，刷新页面可见数据完整持久化在 todos.db 中。

2. 安全边界：严格限制 127.0.0.1

3. 操作系统级开机常驻守护

通过终端手动启动只能供调试使用。若要让待办系统如系统原生工具般随时待命，应通过操作系统的守护进程服务进行管理。

方案 A：macOS 环境下使用 launchd 守护

在用户目录下的 ~/Library/LaunchAgents/ 中创建文件 com.user.todo.plist（注意替换其中的绝对路径与用户名）：

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.user.todo</string>
    <key>ProgramArguments</key>
    <array>
        <string>/usr/bin/python3</string>
        <string>/Users/username/todo_project/server.py</string>
    </array>
    <key>WorkingDirectory</key>
    <string>/Users/username/todo_project</string>
    <key>RunAtLoad</key>
    <true/>
    <key>KeepAlive</key>
    <true/>
    <key>StandardOutPath</key>
    <string>/Users/username/todo_project/todo_stdout.log</string>
    <key>StandardErrorPath</key>
    <string>/Users/username/todo_project/todo_stderr.log</string>
</dict>
</plist>

加载并启动该服务：

launchctl load ~/Library/LaunchAgents/com.user.todo.plist

方案 B：Linux 环境下使用 systemd 用户级单元

在用户目录下建立单元文件 ~/.config/systemd/user/todo.service：

[Unit]
Description=Personal Todo System Local Service
After=network.target

[Service]
Type=simple
WorkingDirectory=%h/todo_project
ExecStart=/usr/bin/python3 %h/todo_project/server.py
Restart=always
RestartSec=3

[Install]
WantedBy=default.target

启用并即刻运行：

systemctl --user daemon-reload
systemctl --user enable --now todo.service

4. 数据备份与容灾恢复

由于 SQLite 的单文件特性，备份工作轻而易举。利用 SQLite 自带的在线安全备份接口，编写一个一行 Shell 脚本即可实现带事务锁保护的冷热无缝备份：

# 导出当前数据库的完整安全快照（即便服务正在写入也不会损坏）
sqlite3 todos.db ".backup 'todos_backup_$(date +%Y%m%d).db'"

在需要灾难恢复或在新电脑上部署时，只需将备份文件重命名为 todos.db 放置于项目根目录，启动 server.py 即可原样恢复全部数据。

五、联调验证与常见问题排查

1. 接口功能性联调验证

可以通过 curl 快速核验后端 API 的数据回包完整性：

（1）新增待办任务请求：

curl -X POST http://127.0.0.1:8080/api/todos      -H "Content-Type: application/json"      -d '{"title": "验证自动化备份", "priority": "high"}'

预期返回响应：

{"success": true, "message": "创建成功", "id": 1}

（2）查询当前待办列表：

curl -s http://127.0.0.1:8080/api/todos | python3 -m json.tool

2. 常见问题排查与修复

• 报错 OSError: [Errno 48] Address already in use：
  说明 8080 端口已被其他本地服务（如本地代理、开发服务器）占用。可通过终端命令 lsof -i :8080 查询占用进程的 PID 并终止，或者在 server.py 中将 PORT = 8080 更改为未被占用的端口（如 8089、9090）。
• 报错 sqlite3.OperationalError: database is locked：
  在多线程并发写入时可能短暂触发 SQLite 默认的文件排他锁。本教程在 database.py 中已执行 PRAGMA journal_mode=WAL;。WAL（预写日志）模式允许多个读取线程与一个写入线程同时工作互不阻塞，能从根本上消除该报错。
• 任务列表中文乱码或特殊符号转义异常：
  检查前端 fetch() 提交头是否显式声明了 Content-Type: application/json，并在 Python 服务端接收时确保执行 decode("utf-8") 与 json.dumps(..., ensure_ascii=False)。本教程给出的源码已完全规避字符集转码不一致问题。