使用 FastHTML 和 MongoDB 构建实时任务管理器！

任务管理器最烦的地方，不是新增任务，而是任务状态变了，页面还停留在旧数据上。

比如同事刚把一个任务改成了“已完成”，你这边还显示“进行中”，必须手动刷新才能看到变化。功能不复杂，但如果用传统方式做，前端页面、后端接口、状态同步，很容易越写越散。

这次换个思路：用 Python 的 FastHTML 负责页面和交互，MongoDB 存任务数据，再通过 HTMX 的请求机制更新页面。先把实时任务管理器跑起来，没必要一上来就引入一整套前后端分离架构。

一、先把技术栈搭起来

FastHTML 有个特点，页面元素可以直接用 Python 代码生成。对于习惯写后端的开发者来说，不需要先搭 Vue 或 React 项目，再维护一套独立的前端工程。

安装依赖：

pip install python-fasthtml pymongo

MongoDB 本地启动后，先准备好连接配置。这里建议把连接地址放到环境变量里，别直接写死在业务代码中。

import osfrom pymongo import MongoClientfrom fasthtml.common import *mongo = MongoClient(    os.getenv("MONGO_URI", "mongodb://localhost:27017"),    serverSelectionTimeoutMS=3000,)db = mongo["workbench"]tasks = db["tasks"]tasks.create_index([("status", 1), ("created_at", -1)])

索引这一步别省。任务列表通常会按状态过滤，再按创建时间排序。数据少的时候不加索引也能跑，等任务量上来，查询才可能暴露问题。

另外，MongoDB 的 _id 默认是 ObjectId，页面提交过来的任务 ID 通常是字符串，后面更新时要注意类型转换。

二、任务数据别设计得太复杂

任务管理器先解决几个实际问题：创建任务、查看任务、修改状态。

数据结构可以先这样定：

{    "title": "排查订单接口超时",    "status": "todo",    "owner": "dev01",    "created_at": 1792490400}

status 暂时只允许 todo、doing、done 三种状态。

我一般不建议刚开始就给任务表塞进十几个字段。优先把查询条件、状态流转和数据校验做好，后续真有截止时间、优先级、负责人权限，再按业务补。

三、用 FastHTML 直接渲染任务列表

下面写一个最小可用的页面。任务列表由服务端查询 MongoDB 后生成 HTML，浏览器不需要再单独调用一个 JSON 接口，把数据交给前端框架二次渲染。

app, rt = fast_app()def task_rows():    result = []    for task in tasks.find().sort("created_at", -1).limit(100):        task_id = str(task["_id"])        status = task.get("status", "todo")        result.append(            Tr(                Td(task.get("title", "未命名任务")),                Td(task.get("owner", "-")),                Td(status),                Td(                    Button(                        "完成",                        hx_post=f"/tasks/{task_id}/done",                        hx_target="#task-list",                        hx_swap="innerHTML",                    )                ),            )        )    return result@rt("/")def index():    return Titled(        "实时任务管理器",        Table(            Thead(                Tr(                    Th("任务"),                    Th("负责人"),                    Th("状态"),                    Th("操作"),                )            ),            Tbody(*task_rows(), id="task-list"),        ),    )

这里有个细节：hx_target="#task-list" 指向的是表格的 tbody，因此后端响应也必须是对应的表格行内容，而不是重新返回整个页面。

这类细节如果没对齐，页面看起来像是接口成功了，实际 DOM 却更新错了位置。

四、状态更新要落到数据库

接着实现完成任务的接口。不要只在页面上把状态文字改成 done，否则刷新页面后数据就恢复原样了。

from bson import ObjectIdfrom bson.errors import InvalidIdfrom datetime import datetime, timezone@rt("/tasks/{task_id}/done", methods=["POST"])def finish_task(task_id: str):    try:        oid = ObjectId(task_id)    except InvalidId:        return PlainText("任务 ID 无效", status_code=400)    result = tasks.update_one(        {"_id": oid, "status": {"$ne": "done"}},        {"$set": {            "status": "done",            "updated_at": datetime.now(timezone.utc),        }},    )    if result.matched_count == 0:        return PlainText("任务不存在或已完成", status_code=404)    return task_rows()

更新条件里加上状态判断，是为了避免重复点击时反复执行无意义的更新。

注意，这里是演示核心逻辑，完整应用还需要补上用户身份校验、CSRF 防护以及任务归属检查。否则，只要拿到任务 ID，就可能修改不属于自己的任务。

五、真正的实时刷新，还差一步

前面实现的是局部更新：点击按钮后，当前浏览器能立即看到最新状态。

但这还不算多人协作场景下的实时同步。假设 A 把任务标记完成，B 的页面并不会自动知道这件事。

可以先用 HTMX 定时轮询解决，不需要马上引入 WebSocket。把任务列表放进可定时刷新的容器即可：

@rt("/")def index():    return Titled(        "实时任务管理器",        Div(            Table(                Thead(                    Tr(                        Th("任务"),                        Th("负责人"),                        Th("状态"),                        Th("操作"),                    )                ),                Tbody(*task_rows()),            ),            hx_get="/tasks/fragment",            hx_trigger="every 3s",            hx_swap="innerHTML",        ),    )@rt("/tasks/fragment")def task_fragment():    return Table(        Tbody(*task_rows())    )

这段代码表达的是轮询思路，实际落地时要统一容器和响应片段：刷新接口应返回容器内部需要替换的内容，避免多套一层表格。

另外，FastHTML 的具体组件参数和 HTMX 属性写法要结合所用版本验证，不能只看页面能启动就认为交互已经正确。

每隔三秒请求一次，其他用户的状态变化就能在下一次轮询时体现出来。

当然，轮询也有成本。如果几百个用户同时在线，每人每三秒发一次请求，服务端会持续处理大量重复查询。规模扩大后，可以考虑 SSE 或 WebSocket，由服务端在任务状态变化时主动推送事件。

六、上线前别漏了这几个细节

分页。 示例限制了最近 100 条任务，真实业务应使用分页，避免列表无限增长。

索引。 根据实际过滤条件和排序方式设计复合索引，并用 explain() 看执行计划，别凭感觉加索引。

连接管理。 MongoDB 客户端应复用，不要每次请求都创建一个新的 MongoClient。

并发更新。 如果任务状态存在严格的流转规则，应把允许的旧状态写进更新条件，避免两个用户同时操作导致状态覆盖。

实时性。 轮询适合轻量应用；需要低延迟推送时，再考虑事件通知机制。

FastHTML 配合 MongoDB 的好处，是可以把页面、交互和数据访问都放在 Python 体系里，减少项目初期的工程负担。

但也别把“少写前端代码”理解成“系统就没有复杂度了”。任务状态怎么流转、多人修改怎么处理、权限怎么校验，这些问题不会因为换了框架而消失。

先把数据更新和页面刷新这条链路跑通，再根据真实并发量决定是否升级推送方案，通常比一开始就堆一套复杂架构更稳。

以上就是“使用 FastHTML 和 MongoDB 构建实时任务管理器！”的详细内容，想要了解更多Python教程欢迎持续关注编程学习网。