FastHTML + MongoDB 实现团队任务管理：HTMX 局部更新与实时推送方案

做任务管理器，真正烦人的通常不是增删改查，而是任务状态已经变了，页面却还停留在旧数据上。

比如团队里有几个人同时处理任务，A 刚把任务改成“已完成”，B 不刷新就看不到变化。要是再叠加上任务筛选、状态统计和多人协作，前端代码能不膨胀吗？

这类小工具，我更愿意用 Python、FastHTML 和 MongoDB 组合来做。FastHTML 负责页面与交互，MongoDB 保存任务数据，HTMX 负责把服务器返回的 HTML 片段更新到页面上。

有一点需要提前说清楚：HTMX 能让页面局部更新，但不等于真正的实时推送。如果要求多个用户修改任务后，其他人的页面也立即变化，还需要增加 WebSocket 或 SSE。下面先把任务管理的主流程搭起来，再看实时通知怎么接。

一、先把任务数据设计好

任务管理器的数据不复杂，标题、负责人、状态、创建时间基本够用。MongoDB 使用文档存储，后续需要增加优先级、截止时间，也不用大动表结构。

先安装依赖：

pip install python-fasthtml pymongo

MongoDB 可以使用本地实例，也可以使用 MongoDB Atlas。连接信息通过环境变量传入，别把生产环境的数据库地址和密码直接写在代码里。

import os
from datetime import datetime, timezone

from fasthtml.common import *
from pymongo import MongoClient
from bson import ObjectId

client = MongoClient(os.environ["MONGO_URI"])
db = client["workbench"]
tasks = db["tasks"]
tasks.create_index([("status", 1), ("created_at", -1)])

这里顺手给状态和创建时间建了组合索引。任务数量少的时候，查列表确实感觉不出来差别；等任务积累起来，每次都扫描整张集合，延迟就可能慢慢上来。

还有一个细节：任务状态最好固定取值，例如 todo、doing、done，不要让前端随便传中文状态进去。否则后面统计任务数量时，很容易出现同一个意思却存了好几种值的情况。

二、用 FastHTML 完成任务列表和状态修改

FastHTML 的一个特点是直接用 Python 组织页面元素，不需要先搭一套 React 或 Vue 工程。

先定义页面。下面的代码只负责任务列表和新增任务，状态修改也由后端处理，避免把业务逻辑散落在浏览器里。

from fasthtml.common import *

app, rt = fast_app()

def task_row(task):
    task_id = str(task["_id"])
    return Li(
        Div(
            Strong(task["title"]),
            Small(
                f'负责人：{task["owner"]} · '
                f'状态：{task["status"]}'
            ),
        ),
        Select(
            *[
                Option(
                    state,
                    value=state,
                    selected=state == task["status"],
                )
                for state in ("todo", "doing", "done")
            ],
            name="status",
            hx_post=f"/tasks/{task_id}/status",
            hx_trigger="change",
            hx_target=f"#task-{task_id}",
            hx_swap="outerHTML",
        ),
        id=f"task-{task_id}",
    )

@rt("/")
def home():
    rows = list(tasks.find().sort("created_at", -1).limit(100))
    return Html(
        Head(
            Title("团队任务管理器"),
            Script(src="https://unpkg.com/htmx.org"),
        ),
        Body(
            H1("团队任务"),
            Form(
                Input(name="title", placeholder="这次要处理什么？"),
                Input(name="owner", placeholder="负责人"),
                Button("创建任务"),
                hx_post="/tasks",
                hx_target="#task-list",
                hx_swap="afterbegin",
            ),
            Ul(*(task_row(item) for item in rows), id="task-list"),
        ),
    )

这段代码里有个值得注意的地方：状态下拉框修改后，通过 hx_post 请求服务器，服务器返回更新后的任务行，再由 HTMX 替换页面中的对应元素。

前端不必重新请求整张页面，也不用自己拼接 JSON 再更新 DOM。

不过，任务新增接口和状态接口还没有写完。实际项目中，新增任务时需要校验标题和负责人，状态更新时也必须检查任务是否存在、状态是否合法。否则页面做得再漂亮，接口一样可能收到脏数据。

三、MongoDB 更新不能只图省事

状态更新最好使用 MongoDB 的原子更新操作，而不是先把整个文档读出来、修改，再覆盖写回去。

from bson.errors import InvalidId

@rt("/tasks/{task_id}/status")
def change_status(task_id: str, status: str):
    if status not in {"todo", "doing", "done"}:
        return "不支持的任务状态", 400

    try:
        oid = ObjectId(task_id)
    except InvalidId:
        return "任务 ID 不合法", 400

    task = tasks.find_one_and_update(
        {"_id": oid},
        {"$set": {"status": status}},
        return_document=True,
    )

    if task is None:
        return "任务不存在", 404

    return task_row(task)

这里使用 find_one_and_update，直接完成查询与更新，并返回更新后的文档。

多人同时操作时，这比“先查询、再覆盖整个文档”更稳妥。不过，如果业务要求严格控制状态流转，例如已完成任务不能直接退回待办，就还要把原状态等条件放进更新过滤器，不能只校验新状态是否合法。

新增接口也补上：

@rt("/tasks")
def create_task(title: str, owner: str):
    title = title.strip()
    owner = owner.strip()

    if not title or not owner:
        return "任务标题和负责人不能为空", 400

    task = {
        "title": title[:120],
        "owner": owner[:60],
        "status": "todo",
        "created_at": datetime.now(timezone.utc),
    }
    result = tasks.insert_one(task)
    task["_id"] = result.inserted_id
    return task_row(task)

serve()

到这里，一个基础版本就能创建任务、展示列表并修改状态了。生产环境还要补上登录鉴权、CSRF 防护、分页以及错误日志。特别是任务接口，不能因为知道某个任务的 ID，就允许任何人修改它。

四、让多个用户真正看到实时变化

前面的 HTMX 方案只是请求触发更新。用户 A 修改任务后，用户 B 的页面不会自动刷新。

如果需要实时协作，可以在任务更新成功后发布一条事件，再通过 SSE 或 WebSocket 推送给在线客户端。

例如事件可以设计成这样：

event = {
    "type": "task.updated",
    "task_id": str(task["_id"]),
    "status": task["status"],
}

这里不要急着把整个任务文档广播出去。先明确客户端究竟需要哪些字段，通常任务 ID、状态和更新时间就够了。这样可以减少传输量，也避免把内部字段不小心暴露给其他用户。

如果采用 SSE，可以由服务器维护事件流，在任务更新后通知订阅者刷新对应的任务行；如果使用 WebSocket，则适合后续增加在线协作、即时评论等双向交互。

需要注意，单进程内保存的订阅者列表只能覆盖当前进程。服务部署成多个实例后，任务更新发生在实例 A，连接却落在实例 B，就需要 Redis Pub/Sub 之类的消息通道进行跨实例广播。

MongoDB 变更流（Change Streams）也是一个选择，可以监听集合变化，但它依赖副本集或分片集群等部署条件，不能默认认为本地独立实例就能直接使用。

这个任务管理器真正需要花心思的地方，不是把页面写出来，而是把数据更新、状态约束和事件通知分开。FastHTML 能减少前端样板代码，MongoDB 适合灵活存储任务文档；至于实时推送，要根据用户规模和部署方式选择实现方案。

先让任务更新可靠，再让页面实时变化，后续排查问题也会容易不少。

类似轻量 Web 工具的实现思路，也可以在云栈社区翻一翻，不少开发者会分享这类组合方案。