Crea un Task Manager in tempo reale con FastHTML e MongoDB

FastHTML e MongoDB offrono un approccio ad alta velocità e nativo per Python allo sviluppo web moderno. In questo tutorial, costruiremo un’applicazione di task manager reattiva e in tempo reale, dimostrando un ciclo di vita CRUD completo (Create, Read, Update, Delete) all’interno di un unico file Python facile da mantenere. Per fare pratica con MongoDB in Python, ti consiglio il corso Introduction to MongoDB in Python.

Che cos’è FastHTML?

FastHTML è un framework minimalista e ad alte prestazioni costruito sulle fondamenta di FastAPI. Introduce un paradigma HTML “pythonic”, che permette agli sviluppatori di costruire interi frontend usando funzioni Python riutilizzabili invece dei tradizionali template.

Il suo punto di forza è l’integrazione nativa con HTMX. Utilizzando semplici attributi HTML per guidare gli aggiornamenti lato server, HTMX consente di creare esperienze da single-page application senza la complessità di uno stack di build pesante in JavaScript.

Che cos’è MongoDB?

MongoDB è il database NoSQL documentale leader, general-purpose. Il suo schema flessibile e l’uso di documenti BSON simili a JSON lo rendono ideale per lo sviluppo moderno e iterativo.

Per Python usiamo il driver asincrono ufficiale, Motor, che fornisce un’interfaccia non bloccante perfetta per l’architettura orientata alle prestazioni di FastAPI e FastHTML.

Perché questo stack eccelle

Combinare queste tecnologie crea un ambiente di sviluppo estremamente produttivo:

• Type safety con Pydantic e MongoDB: FastHTML sfrutta Pydantic per modellazione e validazione dei dati. Questi modelli si mappano naturalmente sulla struttura documentale di MongoDB, offrendo un’esperienza “code-first” che elimina la necessità di boilerplate pesante di object-relational mapping (ORM).
• Performance async end-to-end: Abbinando Motor al core asincrono di FastHTML, le operazioni sul database non bloccano mai l’event loop. Questo garantisce alta concorrenza e bassa latenza, fondamentali per applicazioni reattive in tempo reale.
• Meno contesto da cambiare: puoi gestire schema del database, logica backend e componenti frontend in un ecosistema Python unificato, aumentando sensibilmente la velocità di delivery.

Setup e connessione

Ci sono alcuni prerequisiti da seguire per continuare con questo tutorial: 

• Python 3.8+
• Istanza MongoDB: un’installazione locale oppure un cluster MongoDB Atlas (consigliato per funzionalità in tempo reale)
• Conoscenze di base: familiarità con i decorator in Python e la struttura base dell’HTML.

Inizializzazione del progetto

Per mostrare l’efficienza di questo stack, implementeremo l’intera applicazione, inclusi configurazione del database, modelli dati e route reattive, in un unico file app.py.

Installazione 

Ci servono il framework FastHTML, Motor (il driver MongoDB asincrono ufficiale) e Uvicorn come server ASGI.

Configurazione della connessione a MongoDB

Usiamo AsyncIOMotorClient per stabilire una connessione non bloccante. Questo garantisce che, mentre l’applicazione aspetta l’I/O del database, possa continuare a processare altre richieste concorrenti.

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

Definizione del modello dati 

In un flusso di lavoro document-oriented, lo schema vive nel codice dell’applicazione. Usiamo Pydantic v2 per colmare il divario tra il ObjectId BSON di MongoDB e le stringhe standard di Python. Questo assicura che ogni documento in entrata o in uscita dal database sia validato secondo i nostri requisiti.

Definiamo una classe personalizzata PyObjectId. Questo è necessario perché Pydantic non sa gestire nativamente il tipo ObjectId di MongoDB. Usando __get_pydantic_core_schema__, indichiamo a Pydantic di trattare questo tipo come stringa in JSON ma di validarlo come oggetto BSON per il 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
    )

Layout e route iniziale 

Il componente di layout è una funzione di ordine superiore riutilizzabile che incapsula le nostre viste. In questo modo, dipendenze essenziali come Tailwind CSS per lo stile e HTMX per l’interattività restano coerenti in tutta l’applicazione.

Usando i componenti in Title Case di FastHTML (come Main e Div), manteniamo una struttura pulita e “pythonic” che rispecchia l’albero 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)
    )

Ora, se esegui l’app, la tua UI apparirà così:

Immagine: screenshot dopo il setup

Non ci sono ancora task. Nella prossima sezione ci occuperemo di aggiungere e aggiornare i task.

Implementazione CRUD completa

Read: visualizzare i task (GET /)

Per mostrare i task, recuperiamo tutti i documenti e li renderizziamo in un componente. Tuttavia, in uno scenario reale, il database potrebbe essere offline. Il nostro TaskList aggiornato gestisce il caso in modo elegante, fornendo suggerimenti per la risoluzione direttamente nella UI.

Recupero dati resiliente 

Usiamo collection.find().to_list(length=None) per recuperare i documenti in modo asincrono. Racchiudendolo in un blocco try/except, possiamo rilevare se MongoDB è disconnesso e fornire all’utente un feedback immediato.

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: aggiunta di un nuovo task

Il flusso di creazione usa un form HTML arricchito con attributi HTMX. Nella versione aggiornata, estraiamo esplicitamente i dati del form dall’oggetto Request e usiamo model_dump per preparare il documento all’inserimento 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')

Se eseguiamo il codice qui sotto nel terminale:

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"

Potrai vedere che questo nuovo task è stato aggiunto:

Immagine: screenshot dopo l’aggiunta di un task

Update: attivare/disattivare il completamento

Per gestire gli aggiornamenti, usiamo una richiesta PATCH. Questo dimostra un “micro-refresh”, in cui solo la singola riga del task viene aggiornata nel database e ri-renderizzata nella UI usando il suo ID specifico.

@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: rimuovere un task

La cancellazione è avviata da HTMX. Una volta che MongoDB conferma l’eliminazione, il server restituisce una risposta Empty(). HTMX interpreta questa risposta vuota come un segnale per rimuovere l’elemento target (div più vicino) dal 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()

Se facciamo…

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

…il task viene eliminato.

Immagine: screenshot dopo l’eliminazione 

Puoi trovare l’intero script su GitHub.

Conclusione

Sfruttando uno stack moderno e incentrato su Python, hai costruito un’app reattiva che evita la complessità tradizionale dei framework JavaScript lato client. Questa architettura offre diversi vantaggi chiave per lo sviluppo web moderno. La sinergia tra l’UI basata su componenti di FastHTML e il modello documentale flessibile di MongoDB ti permette di mantenere logica di business, integrità dei dati e presentazione in un unico ecosistema coerente. Questo approccio “tutto in Python” riduce sensibilmente l’overhead di sviluppo e la complessità del deploy.

Prossimi passi per i lettori

• Autenticazione utente: usa le dependency di FastAPI per limitare le liste di task a utenti specifici, salvando user_id nel documento del task.
• Query avanzate: usa l’aggregation framework di MongoDB o filtri semplici per aggiungere alla tua UI le viste “Attivi” e “Completati”.
• Deployment: esegui il deploy del tuo app.py usando Uvicorn dietro a un reverse proxy NGINX per performance di livello produzione.