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

任务管理器算是 Web 开发里很常见的小项目：输入一个标题，新增任务，点一下标记完成，不需要了再删除。

看起来功能不多，但真把它从头到尾写一遍，会发现里面正好串起了几个很典型的边界：

• 数据库里的 BSON 文档，怎样变成 Python 对象；
• Python 对象，怎样变成服务器返回的 HTML；
• 服务器返回的 HTML，怎样只替换页面中真正需要变化的那一小块；
• 一个用户改了任务以后，另一个已经打开页面的人，能不能马上看到变化。

这篇就用 FastHTML、MongoDB、Pydantic 和 HTMX，把这条数据流完整走一遍：任务从浏览器发起，经过路由、数据库和模型，最后再以 HTML 片段回到页面。

这套组合很适合做小型、服务端驱动的交互应用。FastHTML 用 Python 组件生成页面，MongoDB 保存任务文档，Pydantic 负责边界校验，HTMX 则负责把一次请求返回的 HTML 片段放回 DOM。

不过这里有个概念最好先分清。本文说的“实时”，主要是请求发生以后，页面局部立即更新，还不是多个浏览器之间的主动同步。真要做到后者，还需要 MongoDB Change Streams，再配合 SSE、WebSocket 之类的浏览器推送通道。

先把这两个层次分开，后面的 CRUD 和“实时”就不容易混在一起。

01先看懂这套技术栈各自做什么

FastHTML 的价值，并不是把 Python 变成另一套 React，而是把页面组件、路由处理和 Python 逻辑放到同一个开发语境里。

可以直接用 Div、Form、Button 这样的 Python 组件表达 HTML 结构，用路由函数处理请求，再返回一个完整页面或 HTML 片段。页面交互则交给 HTMX 的属性来完成，例如：

• 表单提交时发起 POST；
• 点击按钮时发起 PATCH 或 DELETE；
• 指定哪个元素接收返回结果；
• 决定返回结果是替换元素、替换内部内容，还是删除目标节点。

MongoDB 在这里承担的是文档存储。一个任务可以是这样的 BSON 文档：

{
"_id": "ObjectId(...)",
"title": "完成 FastHTML 集成",
"description": "确认任务可以创建、更新和删除",
"completed": false
}

这里不要把 Pydantic 当成数据库本身的 schema migration 工具。它更像是应用层的边界适配器：数据进入 Python 时检查字段，写回 MongoDB 时再把应用里的字段转换成数据库需要的形状。

整个请求链可以先压缩成一句话：

浏览器请求
  -> FastHTML 路由
  -> 异步 MongoDB 查询
  -> Pydantic 校验/转换
  -> Python 组件生成 HTML
  -> HTMX 替换目标 DOM

这条链里，每一层都有自己的职责。真正容易出问题的地方，往往不是某个 API 不会写，而是把一层该做的事推给了另一层。

02先把连接层和任务模型立住

先用 Motor 的异步客户端把连接层写出来，把数据访问的结构搭清楚。实际新建项目时，最好再根据当前 MongoDB Python driver 文档核对异步 API 和迁移路径，不要把某一个驱动类名当成跨版本不变的合同。

import os

from bson import ObjectId
from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import BaseModel, ConfigDict, Field


MONGO_URI = os.environ["MONGO_URI"]
DB_NAME = "fasthtml_tasks_db"

client = AsyncIOMotorClient(MONGO_URI)
db = client[DB_NAME]
tasks = db["tasks"]


class Task(BaseModel):
id: ObjectId | None = Field(default=None, alias="_id")
    title: str
    description: str | None = None
    completed: bool = False

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        populate_by_name=True,
    )

这里有三个点值得先记住。

第一，连接字符串来自环境变量。Atlas 密钥或本地连接信息不要硬编码进 app.py，否则密钥泄露和环境切换的问题会直接跟着代码进入项目。

第二，MongoDB 的 _id 和应用里的 id 是同一个身份字段，只是命名习惯不同。alias="_id" 让 Pydantic 模型可以用更适合 Python 代码的 id，同时保留写回 MongoDB 时的字段名。

第三，ObjectId 的转换并不只是类型注解。路由收到的任务 ID 通常来自 URL，是字符串；查询 MongoDB 时要先确认它是合法的 ObjectId。否则，错误会发生在数据库查询的深处，最后只剩一个很难解释的 500。

实际写路由时，可以再包一层转换函数，把失败尽量提前变成明确的 400 或 404：

from fastapi import HTTPException


def parse_task_id(raw_id: str) -> ObjectId:
if not ObjectId.is_valid(raw_id):
raise HTTPException(status_code=400, detail="Invalid task id")
return ObjectId(raw_id)

这个函数看起来很小，却把一个重要边界固定了：用户输入不直接进入数据库查询。

03页面先渲染出来，再接 CRUD

页面这一层先保持简单：把任务列表和表单放进同一个布局函数里。这样后面的路由只需要返回页面组件或片段，不必同时处理一大段模板逻辑。

from fasthtml.common import *

app, rt = fast_app()


def layout(*components):
return Main(
        Div(
            H1("FastHTML Task Manager"),
            *components,
            cls="container mx-auto max-w-2xl p-6",
        )
    )


def TaskForm():
return Form(
        Input(name="title", placeholder="Task title", required=True),
        Input(name="description", placeholder="Description"),
        Button("Add task", type="submit"),
        method="post",
        action="/add-task",
        hx_post="/add-task",
        hx_target="#task-list",
        hx_swap="outerHTML",
    )


@rt("/")
async def home():
return layout(
await TaskList(),
        TaskForm(),
    )

这里用 hx_target="#task-list" 指定返回内容交给任务列表元素，再用 hx_swap="outerHTML" 把整个列表节点替换掉。

这样做比“提交表单以后返回 JSON，再由前端找字段、改状态、拼 HTML”少了一层客户端状态同步。代价也很明确：我必须让服务器返回浏览器能直接插入的 HTML，同时保持组件边界和 DOM id 稳定。

任务项和任务列表可以这样拆开：

def TaskItem(task: Task):
    task_id = str(task.id)
    label = "✅ " + task.title if task.completed else task.title

return Div(
        Span(label),
        Button(
"Toggle",
            hx_patch=f"/toggle-task/{task_id}",
            hx_target="closest div",
            hx_swap="outerHTML",
        ),
        Button(
"Delete",
            hx_delete=f"/delete-task/{task_id}",
            hx_target="closest div",
            hx_swap="outerHTML",
        ),
id=f"task-{task_id}",
        cls="flex items-center justify-between border-b p-3",
    )


async def TaskList():
    documents = await tasks.find().to_list(length=None)
    items = [Task(**document) for document in documents]

return Div(
        H2("Current tasks"),
        *(
            [TaskItem(task) for task in items]
or [P("No tasks yet.")]
        ),
id="task-list",
    )

数据库文档先进入 Task，再交给 TaskItem。这样 TaskItem 不需要知道 MongoDB 是怎么查询的，它只负责把一个任务渲染成 HTML。

这个边界很重要：数据访问函数负责拿数据，模型负责校验和转换，组件只负责呈现。

04四个 CRUD 路由，返回范围不要一样

读取：返回完整列表

首页读取列表时，把任务逐个转成 TaskItem。数据量不大时，to_list(length=None) 足够直观。

但这套写法不要直接照搬到大任务表里。真正的实现至少还要补：

• 排序字段；
• 分页或游标；
• completed、用户和时间的索引；
• 只读取页面需要的字段；
• 明确空列表和数据库连接失败的返回状态。

如果 MongoDB 没启动，页面可以给出“数据库连接失败”的提示，但不要把所有异常一把抓住，再直接把 str(e) 返回给用户。连接失败、请求参数错误、任务不存在和未知异常，本来就应该走不同的处理路径。

创建：写入后重绘列表

创建任务时，先从表单读取字段，构造 Pydantic 对象，再转换成 MongoDB 文档：

@rt("/add-task", methods=["POST"])
async def add_task(req: Request):
    form = await req.form()
    title = str(form.get("title") or "").strip()
    description = str(form.get("description") or "").strip() or None

if not title:
return Div(
            P("Title is required", cls="text-red-600"),
id="task-list",
        )

    task = Task(title=title, description=description)
    document = task.model_dump(
        by_alias=True,
        exclude_none=True,
    )
    document.pop("_id", None)

    result = await tasks.insert_one(document)
return await TaskList()

这里让创建动作返回完整的 TaskList，是因为列表里多了一个新节点，直接重新生成列表最容易理解。

但这不意味着每个更新都要重绘整个页面。应用很小时这样做确实简单；数据和交互变复杂以后，返回范围越大，页面闪动、无效渲染，以及并发下的状态覆盖都会更明显。

更新：只替换一行

切换完成状态时，只返回被修改的那一条任务：

@rt("/toggle-task/{task_id}", methods=["PATCH"])
async def toggle_task(task_id: str):
    object_id = parse_task_id(task_id)
    current = await tasks.find_one({"_id": object_id})

if current is None:
raise HTTPException(status_code=404, detail="Task not found")

    next_value = not bool(current.get("completed", False))
await tasks.update_one(
        {"_id": object_id},
        {"$set": {"completed": next_value}},
    )

    updated = await tasks.find_one({"_id": object_id})
return TaskItem(Task(**updated))

按钮的目标是最近的任务 div，所以服务器返回的 TaskItem 会把这一行替换掉。其他任务不需要重新查询，也不需要重新拼装。

这里还留着一个并发边界：先读再写很适合用来说明流程，但不是最严谨的并发更新策略。如果多个请求同时切换同一任务，就需要进一步考虑原子更新、版本字段或条件更新。

删除：返回空响应，移除节点

删除时，目标仍然设为当前任务行：

@rt("/delete-task/{task_id}", methods=["DELETE"])
async def delete_task(task_id: str):
    object_id = parse_task_id(task_id)
    result = await tasks.delete_one({"_id": object_id})

if result.deleted_count == 0:
raise HTTPException(status_code=404, detail="Task not found")

return Empty()

当 HTMX 将空响应替换到目标元素时，目标节点就会从页面中消失。这里的“实时感”来自一个很小但完整的闭环：

1. 用户点击删除；
2. 浏览器发送请求；
3. 服务器删除数据库文档；
4. 服务器返回空片段；
5. 浏览器移除当前 DOM 节点。

它没有让浏览器持续监听数据库，也没有让其他客户端自动收到通知。

05“响应式”不等于“多客户端实时”

到这里最容易被“实时”两个字带偏。

更准确地说，当前实现属于请求驱动的局部刷新：请求成功并返回 HTML 片段后，页面会更新对应的 DOM；但如果用户 A 在另一个浏览器里创建了任务，用户 B 的页面并不会因为数据库发生变化就自动刷新。

真正的多客户端同步至少要多出两层：

MongoDB Change Streams
  -> 服务端事件处理
  -> SSE / WebSocket / 其他推送通道
  -> 浏览器接收事件
  -> HTMX 或客户端代码更新 DOM

MongoDB Change Streams 可以监听集合、数据库或部署范围内的变更事件。它适合做事件源，但它不会替你完成以下事情：

• 哪个用户有权看到这条变更；
• 浏览器断线以后从哪里继续；
• 同一事件重复到达时怎么去重；
• 多个事件同时到达时如何保证顺序；
• 浏览器收到事件后更新整张列表还是某一行。

所以更合适的做法，是按需求分三档处理：

如果现在只是验证任务模型和 CRUD 行为，第一档已经足够。过早引入推送，反而会让连接生命周期、鉴权和失败恢复先抢走主要精力。

06从单文件示例走向工程代码

示例阶段可以先把配置、模型、组件和路由都放进一个 app.py，这样最容易顺着一个文件看完整个闭环。

如果项目准备长期使用，再按职责拆分：

app/
├── config.py       # 环境变量和配置模型
├── db.py           # MongoDB client、集合和索引
├── models.py       # Pydantic 模型、ObjectId 转换
├── queries.py      # 查询和写入函数
├── components.py   # TaskList、TaskItem、TaskForm
├── routes.py       # GET/POST/PATCH/DELETE
└── tests/
    ├── test_models.py
    ├── test_queries.py
    └── test_routes.py

还有几项工程补丁，不建议因为“先跑起来”就一直往后拖。

分页和索引。find().to_list(length=None) 对演示友好，对无上限的任务列表不友好。列表按用户、完成状态和更新时间查询时，要让查询条件和索引一起设计。

鉴权。 任务文档里至少要有 user_id 或项目范围字段。否则你只是在做一个所有人共享的任务板，不是多用户任务管理器。

错误分类。 无效的 ObjectId 是请求问题，任务不存在是资源问题，MongoDB 连接失败是依赖服务问题，未知异常则应该进入服务端日志。它们不应该都变成一个红色的 Error: ...。

依赖锁定。 这里先用 Motor 说明异步连接，但当前 MongoDB Python driver 文档也提供了异步客户端方向。因此，实际项目里要在依赖文件中明确选择 Motor 还是 PyMongo Async，并检查 FastHTML、Pydantic 和驱动版本之间的兼容关系，而不是只复制一段 import 就开始部署。

测试返回片段。 这类应用的测试重点不只是“数据库插入成功”，还包括：创建后返回的 HTML 是否包含新任务，切换后只返回目标行，删除后是否返回预期的空响应，错误 ID 是否得到 400，找不到任务是否得到 404。

07什么时候值得继续加实时能力

如果是一个类似的小项目，我更倾向于先把请求驱动的局部刷新跑顺，再根据真实需求判断是否引入 Change Streams。

只有当下面这些条件同时出现时，主动推送才值得进入主线：

• 多个客户端同时打开同一组任务；
• 一个客户端的修改必须在其他客户端无操作时出现；
• 团队能明确事件的权限过滤和数据范围；
• 已经考虑断线重连、重复事件、事件顺序和页面重新同步；
• 数据库部署满足 Change Streams 的运行条件：MongoDB 需要运行在 replica set 或 sharded cluster 上，standalone 部署不支持这条能力。

如果这些条件还没有同时出现，先把 CRUD、模型边界、错误状态和查询性能做好，收益通常更直接。

这套练习真正有价值的地方，也不在于一次把功能堆满，而是先让一个小应用完成从数据库文档到浏览器 DOM 的完整往返。等这个闭环稳定以后，再决定要不要继续把数据库里的变化主动推到浏览器。