FastHTML + MongoDB Task Manager: CRUD, HTMX Partial Updates & Real-Time Distinction

Understanding the Tech Stack

FastHTML keeps page components, routing, and Python logic in a single development context. You express HTML structure with Python components like Div, Form, Button, handle requests with route functions, and return either full pages or HTML fragments. HTMX attributes drive interactions: hx_post, hx_patch, hx_delete for requests; hx_target to select the element that receives the response; hx_swap to decide whether to replace the element, its inner content, or remove it.

MongoDB stores task documents as BSON. A typical document looks like:

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

Pydantic acts as an application-layer boundary adapter: it validates incoming data and converts application fields to the shape MongoDB expects. It is not a database schema migration tool.

The entire request chain compresses to:

Browser request
 -> FastHTML route
 -> Async MongoDB query
 -> Pydantic validation/conversion
 -> Python component generates HTML
 -> HTMX replaces target DOM

Setting Up the Connection Layer and Task Model

Use Motor's async client to establish the connection. Keep the URI in an environment variable to avoid leaking secrets and to ease environment switching.

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,
    )

Three key points:

Connection string comes from the environment; never hard‑code it.

MongoDB's _id and the application's id represent the same identity. alias="_id" lets the model use id in Python while preserving the MongoDB field name on write. ObjectId conversion is not just a type annotation. Route parameters arrive as strings; they must be validated before querying. A helper function converts early and returns clear 400/404 errors:

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)

This fixes an important boundary: user input never reaches the database query directly.

Rendering the Page and Wiring CRUD

Keep the page layer simple: a layout function that composes the task list and the form. Routes then return only the components or fragments they need.

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(),
    )

The form targets #task-list and uses hx_swap="outerHTML" to replace the entire list node. This avoids a client‑side state sync layer, but requires the server to return ready‑to‑insert HTML and to keep component boundaries and DOM IDs stable.

Task items and the list are separate components:

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",
    )

Database documents are first converted to Task models, then passed to TaskItem. This separation ensures data access functions only fetch data, models handle validation and conversion, and components only handle presentation.

Four CRUD Routes with Different Response Scopes

Read: Return Full List

The home route returns the full TaskList. For small datasets to_list(length=None) is fine, but production code must add sorting, pagination or cursors, indexes on completed, user, and time, projection of only needed fields, and distinct handling for empty lists vs. connection failures.

Create: Re-render List After Insert

@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()

Returning the full list is simple for a tiny app, but as data and interactions grow, larger response scopes cause flicker, wasted rendering, and potential race conditions.

Update: Replace Only the Changed Row

@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))

The button targets the closest div, so the returned TaskItem replaces only that row. Other tasks are untouched. Note the concurrency caveat: read‑then‑write works for demonstration but is not a safe concurrent update strategy; production would need atomic updates, version fields, or conditional writes.

Delete: Return Empty Response to Remove Node

@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 replaces the target element with an empty response, removing the node from the DOM. The "real-time feel" comes from a tight request‑response loop:

User clicks delete.

Browser sends request.

Server deletes the document.

Server returns empty fragment.

Browser removes the current DOM node.

No continuous database listening, no automatic notification to other clients.

"Responsive" Does Not Mean Multi‑Client Real‑Time

The current implementation is request‑driven partial refresh : the page updates after a successful request returns HTML. If user A creates a task in another browser, user B's page does not auto‑refresh.

True multi‑client sync adds at least two layers:

MongoDB Change Streams
 -> Server-side event handling
 -> SSE / WebSocket / other push channels
 -> Browser receives event
 -> HTMX or client code updates DOM

Change Streams emit change events but do not solve:

Which user may see a given change.

Resuming after a disconnect.

Deduplicating repeated events.

Ordering simultaneous events.

Whether to refresh the whole list or a single row.

A pragmatic three‑tier approach:

Single‑user, small internal tool: HTMX requests + server HTML fragments. No Change Streams, no WebSockets.

Multi‑user, manual refresh acceptable: HTMX partial refresh + query filtering. No persistent push connections.

Multiple clients must sync actively: Change Streams + SSE/WebSocket + event handling. HTMX requests alone are insufficient.

If you only need to validate the task model and CRUD behavior, tier one is enough. Prematurely adding push shifts focus to connection lifecycle, auth, and failure recovery.

Moving from a Single‑File Example to Production Code

Start with everything in app.py to see the full loop. For long‑term projects, split by responsibility:

app/
├── config.py         # Environment variables and config models
├── db.py             # MongoDB client, collections, indexes
├── models.py         # Pydantic models, ObjectId conversion
├── queries.py        # Query and write functions
├── components.py     # TaskList, TaskItem, TaskForm
├── routes.py         # GET/POST/PATCH/DELETE
└── tests/
    ├── test_models.py
    ├── test_queries.py
    └── test_routes.py

Engineering patches not to defer:

Pagination and indexes. find().to_list(length=None) is demo‑friendly but unsafe for unbounded lists. Design queries and indexes together for user, completion status, and update time.

Authentication. Task documents need a user_id or project scope field; otherwise it's a shared board, not a multi‑user manager.

Error classification. Invalid ObjectId = 400, missing task = 404, DB connection failure = 503, unknown errors = logged server‑side. Don't collapse all into a generic red error message.

Dependency locking. Motor is used here, but the official MongoDB Python driver now offers an async client. Pin Motor or PyMongo Async in dependencies and verify compatibility with FastHTML and Pydantic versions.

Test HTML fragments. Tests should verify: create returns HTML containing the new task, toggle returns only the target row, delete returns an empty response, bad IDs yield 400, missing tasks yield 404.

When to Add Real‑Time Capabilities

Prefer to get request‑driven partial refresh working first, then introduce Change Streams only when all these conditions appear together:

Multiple clients simultaneously view the same task set.

One client's changes must appear on others without user action.

The team can define permission filtering and data scope for events.

Disconnect/reconnect, duplicate events, event ordering, and page re‑sync are already designed.

The MongoDB deployment runs as a replica set or sharded cluster (standalone does not support Change Streams).

If those conditions aren't met, investing in CRUD correctness, model boundaries, error states, and query performance yields more immediate value. The real benefit of this exercise is completing the full round‑trip from database document to browser DOM; once that loop is stable, you can decide whether to push database changes to the browser proactively.