Créer un gestionnaire de tâches en temps réel avec FastHTML et MongoDB

FastHTML et MongoDB offrent une approche rapide et native Python du développement web moderne. Dans ce tutoriel, nous allons créer une application de gestion de tâches réactive et temps réel, illustrant un cycle CRUD complet (Create, Read, Update, Delete) au sein d’un unique fichier Python, simple à maintenir. Pour vous exercer à l’usage de MongoDB en Python, nous vous recommandons le cours Introduction to MongoDB in Python.

Qu’est-ce que FastHTML ?

FastHTML est un framework minimaliste et performant, bâti sur les fondations de FastAPI. Il introduit un paradigme HTML « pythonique », permettant de construire tout le frontend avec des fonctions Python réutilisables plutôt qu’avec des templates traditionnels.

Sa force réside dans son intégration native avec HTMX. En utilisant de simples attributs HTML pour piloter les mises à jour côté serveur, HTMX permet de créer des expériences proches des applications monopage, sans la complexité d’une chaîne de build lourde en JavaScript.

Qu’est-ce que MongoDB ?

MongoDB est la base de données NoSQL généraliste et orientée documents de référence. Son schéma flexible et l’utilisation de documents BSON proches de JSON en font un excellent choix pour un développement moderne et itératif.

En Python, nous utilisons le driver asynchrone officiel, Motor, qui propose une interface non bloquante parfaitement adaptée à l’architecture performante de FastAPI et FastHTML.

Pourquoi cette pile brille

La combinaison de ces technologies crée un environnement de développement d’une productivité remarquable :

• Sécurité de type avec Pydantic et MongoDB : FastHTML s’appuie sur Pydantic pour la modélisation et la validation des données. Ces modèles s’alignent naturellement sur la structure documentaire de MongoDB, offrant une expérience « code-first » qui évite le boilerplate lourd des ORM.
• Performance asynchrone de bout en bout : En associant Motor au cœur asynchrone de FastHTML, les opérations sur la base ne bloquent jamais la boucle d’événements. Vous gagnez en concurrence et en latence, deux critères clés pour des applications réactives et temps réel.
• Moins de changement de contexte : Vous gérez schéma, logique backend et composants frontend dans un écosystème Python unifié, ce qui accélère fortement la livraison.

Configuration et connexion

Quelques prérequis sont nécessaires pour suivre ce tutoriel :

• Python 3.8+
• Instance MongoDB : une installation locale ou un cluster MongoDB Atlas (recommandé pour les fonctionnalités temps réel)
• Connaissances de base : être à l’aise avec les décorateurs Python et la structure HTML.

Initialisation du projet

Pour démontrer l’efficacité de cette pile, nous allons implémenter l’intégralité de l’application, y compris la configuration de la base, les modèles de données et les routes réactives, dans un seul fichier app.py.

Installation 

Nous avons besoin du framework FastHTML, de Motor (le driver MongoDB asynchrone officiel) et d’Uvicorn pour le serveur ASGI.

Configuration de la connexion MongoDB

Nous utilisons AsyncIOMotorClient pour établir une connexion non bloquante. Ainsi, pendant que l’application attend des E/S base de données, elle continue à traiter d’autres requêtes concurrentes.

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

Définir le modèle de données 

Dans un flux orienté documents, le schéma vit dans votre code applicatif. Nous utilisons Pydantic v2 pour faire le lien entre l’ObjectId BSON de MongoDB et de simples chaînes de caractères Python. Ainsi, chaque document entrant ou sortant de la base est validé selon nos exigences.

Nous définissons une classe personnalisée PyObjectId. C’est nécessaire car Pydantic ne sait pas gérer nativement le type ObjectId de MongoDB. Avec __get_pydantic_core_schema__, nous indiquons à Pydantic de traiter ce type comme une chaîne en JSON mais de le valider comme un objet BSON pour la base.

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
    )

Mise en page et route initiale 

Le composant de mise en page est une fonction de plus haut niveau et réutilisable qui encapsule nos vues. Elle garantit que les dépendances essentielles, comme Tailwind CSS pour le style et HTMX pour l’interactivité, sont homogènes dans toute l’application.

En utilisant les composants en « Title Case » de FastHTML (comme Main et Div), nous conservons une structure pythonique claire, fidèle à l’arborescence HTML.

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

À ce stade, si vous lancez l’application, l’interface ressemblera à ceci :

Image : capture d’écran après la configuration

Il n’y a pas encore de tâches. Nous allons désormais travailler sur l’ajout et la mise à jour de tâches.

Implémentation CRUD complète

Read : afficher les tâches (GET /)

Pour afficher les tâches, nous récupérons tous les documents et les rendons dans un composant. Dans un contexte réel, la base peut toutefois être hors ligne. Notre TaskList mis à jour gère cela proprement en proposant, directement dans l’interface, des étapes de diagnostic.

Récupération de données résiliente 

Nous utilisons collection.find().to_list(length=None) pour récupérer les documents de manière asynchrone. En l’encapsulant dans un bloc try/except, nous détectons une éventuelle déconnexion de MongoDB et informons immédiatement l’utilisateur.

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 : ajouter une nouvelle tâche

La création s’appuie sur un formulaire HTML enrichi avec des attributs HTMX. Dans cette version, nous extrayons explicitement les données du formulaire à partir de l’objet Request et utilisons model_dump pour préparer le document à l’insertion dans 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')

Si nous exécutons la commande suivante dans le terminal :

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"

Vous verrez que la nouvelle tâche est bien ajoutée :

Image : capture d’écran après l’ajout d’une tâche

Update : basculer l’état d’achèvement

Pour gérer les mises à jour, nous utilisons une requête PATCH. Cela illustre un « micro-refresh » : seule la ligne de tâche concernée est mise à jour en base, puis re-rendue dans l’UI à partir de son ID.

@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 : supprimer une tâche

La suppression est déclenchée par HTMX. Une fois la suppression confirmée par MongoDB, le serveur renvoie une réponse Empty(). HTMX interprète cette réponse vide comme un signal pour retirer l’élément ciblé (closest div) du DOM.

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

Si nous exécutons…

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

… la tâche est supprimée.

Image : capture d’écran après suppression 

Vous pouvez consulter le script complet sur GitHub.

Conclusion

En tirant parti d’une pile moderne et centrée sur Python, vous avez construit une application réactive sans la complexité habituelle des frameworks JavaScript côté client. Cette architecture présente plusieurs avantages clés pour le web moderne. La synergie entre l’UI basée composants de FastHTML et le modèle documentaire flexible de MongoDB vous permet de réunir logique métier, intégrité des données et présentation au sein d’un même écosystème. Cette approche « tout en Python » réduit significativement la charge de développement et la complexité de déploiement.

Prochaines étapes

• Authentification : utilisez les dépendances FastAPI pour restreindre les listes de tâches à des utilisateurs spécifiques, en stockant user_id dans le document de tâche.
• Requêtes avancées : utilisez le framework d’agrégation de MongoDB ou de simples filtres pour ajouter des vues « Actives » et « Terminées » à votre interface.
• Déploiement : déployez votre app.py avec Uvicorn derrière un proxy inverse NGINX pour des performances de niveau production.