Bouw een realtime takenmanager met FastHTML en MongoDB

FastHTML en MongoDB bieden een snelle, Python-native aanpak voor moderne webontwikkeling. In deze tutorial bouwen we een reactieve, realtime takenmanager en laten we een complete CRUD-cyclus (Create, Read, Update, Delete) zien binnen één onderhoudbaar Python-bestand. Wil je praktisch aan de slag met MongoDB in Python? Dan raad ik de cursus Introduction to MongoDB in Python aan.

Wat is FastHTML?

FastHTML is een minimalistisch, high-performance framework gebouwd op de fundamenten van FastAPI. Het introduceert een Pythonische HTML-paradigma, waardoor je volledige frontends kunt bouwen met herbruikbare Python-functies in plaats van traditionele templates.

De kernkracht zit in de native integratie met HTMX. Door eenvoudige HTML-attributen te gebruiken voor server-side updates, kun je met HTMX dynamische single-page-ervaringen creëren zonder de complexiteit van een JavaScript-zware buildstack.

Wat is MongoDB?

MongoDB is de toonaangevende, general-purpose, documentgebaseerde NoSQL-database. De flexibele schema’s en het gebruik van JSON-achtige BSON-documenten maken het ideaal voor moderne, iteratieve ontwikkeling.

Voor Python gebruiken we de officiële asynchrone driver Motor, die een niet-blokkerende interface biedt die perfect past bij de performancegerichte architectuur van FastAPI en FastHTML.

Waarom deze stack uitblinkt

Door deze technologieën te combineren, ontstaat een uitzonderlijk productieve ontwikkelomgeving:

• Typeveiligheid met Pydantic en MongoDB: FastHTML gebruikt Pydantic voor datamodellering en validatie. Deze modellen sluiten naadloos aan op de documentstructuur van MongoDB en bieden een “code-first”-ervaring die de noodzaak van zware ORM-boilerplate wegneemt.
• End-to-end async performance: Door Motor te koppelen aan de asynchrone kern van FastHTML blokkeren database-operaties nooit de eventloop. Dit zorgt voor hoge gelijktijdigheid en lage latency, cruciaal voor realtime, reactieve applicaties.
• Minder contextswitching: Je kunt het databaseschema, de backendlogica en de frontendcomponenten beheren binnen één Python-ecosysteem, wat de opleversnelheid aanzienlijk verhoogt.

Setup en verbinding

Er zijn een paar vereisten om deze tutorial te volgen: 

• Python 3.8+
• MongoDB-instantie: een lokale installatie of een MongoDB Atlas-cluster (aanbevolen voor realtime features)
• Basiskennis: bekendheid met Python-decorators en de basis HTML-structuur.

Projectinitialisatie

Om de efficiëntie van deze stack te tonen, implementeren we de volledige applicatie, inclusief databaseconfiguratie, datamodellen en reactieve routes, in één app.py-bestand.

Installatie 

We hebben het FastHTML-framework, Motor (de officiële asynchrone MongoDB-driver) en Uvicorn voor de ASGI-server nodig.

MongoDB-verbinding instellen

We gebruiken AsyncIOMotorClient om een niet-blokkerende verbinding op te zetten. Zo kan je applicatie andere gelijktijdige verzoeken blijven verwerken terwijl er op database-I/O wordt gewacht.

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

Het datamodel definiëren 

In een documentgeoriënteerde workflow leeft het schema in je applicatiecode. We gebruiken Pydantic v2 om de kloof te overbruggen tussen MongoDB’s BSON-ObjectId en standaard Python-strings. Zo wordt elk document dat onze database in- of uitgaat gevalideerd tegen onze eisen.

We definiëren een aangepaste PyObjectId-klasse. Dit is nodig omdat Pydantic het MongoDB-ObjectId-type niet standaard begrijpt. Met __get_pydantic_core_schema__ vertellen we Pydantic dit type als string in JSON te behandelen maar het als BSON-object te valideren voor de database.

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
    )

Lay-out en initiële route 

De lay-outcomponent is een herbruikbare higher-orderfunctie die onze views omwikkelt. Zo blijven essentiële afhankelijkheden, zoals Tailwind CSS voor styling en HTMX voor interactiviteit, consistent in de hele applicatie.

Door de title-case-componenten van FastHTML te gebruiken (zoals Main en Div) behouden we een schone, Pythonische structuur die de HTML-boom weerspiegelt.

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

Als je de app nu draait, ziet je UI er zo uit:

Afbeelding: screenshot na setup

Er zijn nog geen taken. In de volgende sectie gaan we taken toevoegen en bijwerken.

Volledige CRUD-implementatie

Read: Taken weergeven (GET /)

Om taken weer te geven, halen we alle documenten op en renderen die in een component. In een echte omgeving kan de database echter offline zijn. Onze bijgewerkte TaskList gaat hier netjes mee om door in de UI direct stappen voor probleemoplossing te tonen.

Robuust data ophalen 

We gebruiken collection.find().to_list(length=None) om documenten asynchroon op te halen. Door dit in een try/except te wikkelen, kunnen we detecteren of MongoDB is losgekoppeld en de gebruiker meteen feedback geven.

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: Een nieuwe taak toevoegen

De creatiestroom gebruikt een HTML-formulier met HTMX-attributen. In de bijgewerkte versie halen we formulierdata expliciet uit het Request-object en gebruiken we model_dump om het document voor te bereiden voor invoeging in MongoDB.

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

Als we onderstaande code in de terminal draaien:

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"

Dan zie je dat deze nieuwe taak is toegevoegd:

Afbeelding: screenshot na het toevoegen van een taak

Update: Voltooiing toggelen

Voor updates gebruiken we een PATCH-verzoek. Dit demonstreert een “micro-refresh”, waarbij alleen de enkele taakrij in de database wordt bijgewerkt en met zijn specifieke ID opnieuw in de UI wordt gerenderd.

@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: Een taak verwijderen

Verwijderen wordt gestart door HTMX. Zodra MongoDB de verwijdering bevestigt, retourneert de server een Empty()-response. HTMX interpreteert deze lege response als een signaal om het doel-element (dichtstbijzijnde div) uit de DOM te verwijderen.

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

Als we…

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

…doen, wordt de taak verwijderd.

Afbeelding: screenshot na verwijderen 

Je vindt het volledige script op GitHub.

Conclusie

Door een moderne, Python-centrische stack te benutten, heb je een reactieve applicatie gebouwd die de traditionele complexiteit van client-side JavaScript-frameworks vermijdt. Deze specifieke architectuur biedt verschillende belangrijke voordelen voor moderne webontwikkeling. De synergie tussen de componentgebaseerde UI van FastHTML en het flexibele documentmodel van MongoDB stelt je in staat businesslogica, dataintegriteit en presentatie te behouden binnen één samenhangend ecosysteem. Deze “alles-in-Python”-aanpak vermindert de ontwikkeloverhead en uitrolcomplexiteit aanzienlijk.

Volgende stappen voor lezers

• Gebruikersauthenticatie: Gebruik FastAPI-dependencies om takenlijsten te beperken tot specifieke gebruikers, en sla user_id op in het taakdocument.
• Geavanceerde queries: Gebruik MongoDB’s aggregatiekader of eenvoudige filters om “Actief”- en “Voltooid”-weergaven aan je UI toe te voegen.
• Deployen: Zet je app.py uit met Uvicorn achter een NGINX reverse proxy voor performance op productieniveau.