Erstelle einen Echtzeit-Task-Manager mit FastHTML und MongoDB

FastHTML und MongoDB ermöglichen einen schnellen, Python‑nativen Ansatz für moderne Webentwicklung. In diesem Tutorial bauen wir eine reaktive Task-Manager-App in Echtzeit und demonstrieren einen vollständigen CRUD‑Lebenszyklus (Create, Read, Update, Delete) in einer einzigen, gut wartbaren Python-Datei. Für praktische Erfahrung mit MongoDB in Python empfehle ich den Kurs Introduction to MongoDB in Python.

Was ist FastHTML?

FastHTML ist ein minimalistisches, leistungsstarkes Framework auf Basis von FastAPI. Es führt ein pythonisches HTML-Paradigma ein, mit dem sich komplette Frontends mit wiederverwendbaren Python-Funktionen statt klassischer Templates entwickeln lassen.

Seine größte Stärke ist die native Integration mit HTMX. Mit einfachen HTML-Attributen für serverseitige Updates ermöglicht HTMX dynamische Single-Page-Erlebnisse – ohne die Komplexität eines JavaScript‑lastigen Build‑Stacks.

Was ist MongoDB?

MongoDB ist die führende, dokumentenbasierte NoSQL-Datenbank für allgemeine Anwendungsfälle. Das flexible Schema und die JSON‑ähnlichen BSON‑Dokumente machen sie ideal für moderne, iterative Entwicklung.

Für Python nutzen wir den offiziellen asynchronen Treiber Motor, der eine nicht blockierende Schnittstelle bietet – perfekt für die leistungsorientierte Architektur von FastAPI und FastHTML.

Warum dieser Stack überzeugt

Die Kombination dieser Technologien schafft eine äußerst produktive Entwicklungsumgebung:

• Typensicherheit mit Pydantic und MongoDB: FastHTML nutzt Pydantic für Datenmodellierung und ‑validierung. Diese Modelle passen sich natürlich an die Dokumentstruktur von MongoDB an und bieten ein „Code‑first“-Erlebnis ohne umfangreichen ORM‑Overhead.
• End-to-end asynchrone Performance: In Kombination aus Motor und dem asynchronen Kern von FastHTML blockieren Datenbankoperationen nie die Event‑Loop. Das sorgt für hohe Parallelität und geringe Latenz – entscheidend für reaktive Echtzeitanwendungen.
• Weniger Kontextwechsel: Du verwaltest Datenbankschema, Backend-Logik und Frontend‑Komponenten in einem einheitlichen Python‑Ökosystem – das beschleunigt die Auslieferung spürbar.

Setup und Verbindung

Es gibt ein paar Voraussetzungen, um diesem Tutorial zu folgen: 

• Python 3.8+
• MongoDB-Instanz: eine lokale Installation oder ein MongoDB Atlas‑Cluster (für Echtzeitfunktionen empfohlen)
• Grundkenntnisse: Vertrautheit mit Python‑Dekoratoren und grundlegender HTML‑Struktur.

Projektinitialisierung

Um die Effizienz dieses Stacks zu zeigen, implementieren wir die gesamte Anwendung – inklusive Datenbankkonfiguration, Datenmodellen und reaktiven Routen – in einer einzigen Datei app.py.

Installation 

Wir benötigen das FastHTML‑Framework, Motor (den offiziellen asynchronen MongoDB‑Treiber) und Uvicorn als ASGI‑Server.

Einrichtung der MongoDB-Verbindung

Wir verwenden AsyncIOMotorClient für eine nicht blockierende Verbindung. So kann deine App während der Wartezeit auf Datenbank‑I/O andere Anfragen parallel weiterverarbeiten.

import os
from motor.motor_asyncio import AsyncIOMotorClient
from fasthtml.common import *
from bson import ObjectId
from pydantic import Field, BaseModel
from pymongo.errors import ServerSelectionTimeoutError, ConnectionFailure


# configure MongoDB
MONGO_URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017/") # your Mongo URI
DB_NAME = "fasthtml_tasks_db"
COLLECTION_NAME = "tasks"

# Initialize the async client and reference our collections
client = AsyncIOMotorClient(MONGO_URI)
db = client[DB_NAME]
collection = db[COLLECTION_NAME]

# FastHTML Application Initialization
app = FastHTML()

Definieren des Datenmodells 

In einem dokumentenorientierten Workflow lebt das Schema im Anwendungscode. Wir nutzen Pydantic v2, um die Lücke zwischen MongoDBs BSON‑ObjectId und normalen Python‑Strings zu schließen. So wird jedes Dokument, das in die Datenbank geht oder sie verlässt, gegen unsere Anforderungen validiert.

Wir definieren eine eigene PyObjectId‑Klasse. Das ist nötig, weil Pydantic den MongoDB‑ObjectId‑Typ nicht nativ versteht. Mit __get_pydantic_core_schema__ teilen wir Pydantic mit, diesen Typ in JSON als String zu behandeln, ihn für die Datenbank aber als BSON‑Objekt zu validieren.

python
from pydantic import BaseModel, Field
from pydantic_core import core_schema
from pydantic.json_schema import JsonSchemaValue

class PyObjectId(ObjectId):
    """Custom type to bridge MongoDB ObjectId and Pydantic v2 validation."""
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type, handler):
        # Defines how Pydantic should validate this type
        return core_schema.no_info_plain_validator_function(cls.validate)

    @classmethod
    def validate(cls, v):
        if isinstance(v, ObjectId): return v
        if isinstance(v, str) and ObjectId.is_valid(v): return ObjectId(v)
        raise ValueError("Invalid ObjectId")

    @classmethod
    def __get_pydantic_json_schema__(cls, _core_schema, handler) -> JsonSchemaValue:
        # Ensures the OpenAPI/JSON schema reflects this as a string
        return {"type": "string"}

class Task(BaseModel):
    """Pydantic model representing the Task document schema."""
    # Maps MongoDB's internal '_id' to a developer-friendly 'id' field
    id: PyObjectId | None = Field(None, alias='_id')
    title: str
    description: str | None = None
    completed: bool = False

    model_config = ConfigDict(
        arbitrary_types_allowed=True
    )

Layout und Initialroute 

Die Layout‑Komponente ist eine wiederverwendbare Higher‑Order‑Funktion, die unsere Views umschließt. So stellen wir sicher, dass wichtige Abhängigkeiten wie Tailwind CSS für das Styling und HTMX für Interaktivität in der gesamten Anwendung konsistent sind.

Mit den titelgeschriebenen Komponenten von FastHTML (wie Main und Div) behalten wir eine saubere, pythonische Struktur bei, die dem HTML‑Baum entspricht.

# A responsive layout using Tailwind CSS and the HTMX CDN
def layout(*comps):
    """Wraps the application content in a consistent container."""
    return Main(
        Div(
            H1('📝 Real-Time FastHTML Task Manager', 
               cls='text-3xl font-bold mb-8 text-center text-gray-800'),
            *comps,
            cls='container mx-auto p-6 max-w-2xl'
        )
    )

@app.get('/')
async def home():
    """Initial route rendering the core application layout."""
    return layout(
        await TaskList(), # Fetches and displays existing tasks (Read)
        TaskForm()        # Renders the entry form (Create)
    )

Wenn du die App jetzt startest, sieht dein UI so aus:

Bild: Screenshot nach dem Setup

Es gibt noch keine Tasks. Im nächsten Abschnitt fügen wir Aufgaben hinzu und aktualisieren sie.

Vollständige CRUD‑Implementierung

Read: Aufgaben anzeigen (GET /)

Zum Anzeigen der Tasks rufen wir alle Dokumente ab und rendern sie in einer Komponente. In der Praxis kann die Datenbank aber offline sein. Unsere überarbeitete TaskList geht damit souverän um und zeigt direkt in der UI Hinweise zur Fehlerbehebung.

Robustes Daten-Fetching 

Wir nutzen collection.find().to_list(length=None) für den asynchronen Dokumentabruf. In einem try/except‑Block erkennen wir Verbindungsprobleme zu MongoDB und geben sofort Rückmeldung an die Nutzer.

async def TaskList():
    """Fetches documents from MongoDB and hydrates the Task list view with error handling."""
    try:
        tasks_data = await collection.find().to_list(length=None)
        tasks = [Task(**doc) for doc in tasks_data]
    except (ServerSelectionTimeoutError, ConnectionFailure, Exception) as e:
        # Catch MongoDB connection errors and provide troubleshooting tips
        return Div(
            H2('Current Tasks', cls='text-xl font-semibold mb-4'),
            Div(
                P('⚠️ MongoDB is not running.', cls='text-red-600 font-semibold mb-2'),
                P('To start MongoDB:', cls='text-gray-600 mb-1'),
                P('1. brew install mongodb-community', cls='text-gray-500 text-sm ml-4'),
                P('2. brew services start mongodb-community', cls='text-gray-500 text-sm ml-4'),
                cls='p-4 bg-yellow-50 border border-yellow-200 rounded'
            ),
            id='task-list'
        )

    return Div(
        H2('Current Tasks', cls='text-xl font-semibold mb-4'),
        *[TaskItem(task) for task in tasks] if tasks else P('No tasks yet. Add one below!', cls='text-gray-500 italic p-4'),
        id='task-list',
        cls='bg-white rounded-lg shadow-sm border border-gray-200'
    )

Create: Eine neue Aufgabe hinzufügen

Für das Anlegen nutzen wir ein HTML‑Formular mit HTMX‑Attributen. In der aktualisierten Version lesen wir die Formulardaten explizit aus dem Request‑Objekt aus und verwenden model_dump, um das Dokument für das Einfügen in MongoDB aufzubereiten.

@app.post('/add-task')
async def add_task(req: Request):
    """Create: Inserts a task and returns the refreshed list."""
    try:
        form_data = await req.form()
        task = Task(
            title=form_data.get('title', ''),
            description=form_data.get('description') or None
        )
        
        # model_dump(by_alias=True) ensures 'id' is converted back to '_id' for Mongo
        task_dict = task.model_dump(exclude_none=True, by_alias=True)
        
        # Remove _id if it's None so MongoDB can generate its own unique ID
        if '_id' in task_dict and task_dict['_id'] is None:
            del task_dict['_id']
            
        await collection.insert_one(task_dict)
        return await TaskList()
    except Exception as e:
        return Div(P(f'Error: {str(e)}', cls='text-red-500 p-4'), id='task-list')

Wenn wir folgenden Code im Terminal ausführen:

curl -X POST http://localhost:8000/add-task -d "title=Complete FastHTML MongoDB integration" -d "description=Verify that tasks can be created, updated, and deleted successfully" -H "Content-Type: application/x-www-form-urlencoded"

Dann siehst du, dass die neue Aufgabe hinzugefügt wurde:

Bild: Screenshot nach dem Hinzufügen einer Aufgabe

Update: Abschlussstatus umschalten

Für Updates verwenden wir eine PATCH‑Anfrage. Das demonstriert ein „Mikro‑Refresh“: Nur die einzelne Task‑Zeile wird in der Datenbank aktualisiert und über ihre spezifische ID im UI neu gerendert.

@app.patch('/toggle-task/{task_id}')
async def toggle_task(task_id: str):
    """Update: Toggles completion status and returns the single row fragment."""
    task_doc = await collection.find_one({"_id": ObjectId(task_id)})
    if not task_doc: raise HTTPException(404, "Task not found")

    await collection.update_one(
        {"_id": ObjectId(task_id)},
        {"$set": {"completed": not task_doc['completed']}}
    )

    # Return only the updated TaskItem for a surgical DOM update
    updated_doc = await collection.find_one({"_id": ObjectId(task_id)})
    return TaskItem(Task(**updated_doc))

Delete: Eine Aufgabe löschen

Das Löschen wird durch HTMX ausgelöst. Sobald MongoDB die Löschung bestätigt, liefert der Server eine Empty()‑Antwort. HTMX interpretiert diese leere Antwort als Signal, das Zielelement (nächstgelegenes div) aus dem DOM zu entfernen.

@app.delete('/delete-task/{task_id}')
async def delete_task(task_id: str):
    """Delete: Removes a task from MongoDB."""
    await collection.delete_one({"_id": ObjectId(task_id)})
    # Signal HTMX to remove the element
    return Empty()

Wenn wir …

curl -X DELETE http://localhost:8000/delete-task/695968244236010c04f313fa

… ausführen, wird die Aufgabe gelöscht.

Bild: Screenshot nach dem Löschen 

Das komplette Skript findest du auf GitHub.

Fazit

Mit einem modernen, Python‑zentrierten Stack hast du eine reaktive Anwendung gebaut – ohne die übliche Komplexität clientseitiger JavaScript‑Frameworks. Diese Architektur bietet mehrere Vorteile für die Webentwicklung: Die Synergie aus der komponentenbasierten UI von FastHTML und dem flexiblen Dokumentenmodell von MongoDB ermöglicht es dir, Geschäftslogik, Datenintegrität und Präsentation in einem kohärenten Ökosystem zu vereinen. Dieser „All‑in‑Python“-Ansatz reduziert Entwicklungsaufwand und Deployment‑Komplexität deutlich.

Nächste Schritte für Leserinnen und Leser

• Nutzer-Authentifizierung: Verwende FastAPI‑Dependencies, um Task‑Listen auf bestimmte Nutzer zu beschränken und speichere die user_id im Task‑Dokument.
• Erweiterte Abfragen: Nutze das MongoDB‑Aggregationsframework oder einfache Filter, um „Aktiv“- und „Erledigt“-Ansichten in deiner UI hinzuzufügen.
• Deployment: Setze deine app.py mit Uvicorn hinter einem NGINX‑Reverse‑Proxy produktionsreif in Betrieb.