Crea un gestor de tareas en tiempo real con FastHTML y MongoDB

FastHTML y MongoDB ofrecen un enfoque veloz y nativo de Python para el desarrollo web moderno. En este tutorial, crearemos una aplicación de gestor de tareas reactiva y en tiempo real, demostrando un ciclo CRUD completo (crear, leer, actualizar, eliminar) dentro de un único y mantenible archivo de Python. Para practicar con MongoDB en Python, te recomiendo el curso Introduction to MongoDB in Python.

¿Qué es FastHTML?

FastHTML es un framework minimalista y de alto rendimiento construido sobre FastAPI. Introduce un paradigma de HTML pythónico que permite crear frontends completos con funciones reutilizables de Python en lugar de plantillas tradicionales.

Su gran fortaleza es la integración nativa con HTMX. Al usar atributos HTML sencillos para activar actualizaciones en el servidor, HTMX te permite crear experiencias de aplicación de una sola página sin la complejidad de un stack cargado de JavaScript.

¿Qué es MongoDB?

MongoDB es la base de datos NoSQL documental y de propósito general líder del mercado. Su esquema flexible y el uso de documentos BSON (similares a JSON) la hacen ideal para el desarrollo moderno e iterativo.

Para Python, usamos el driver asíncrono oficial, Motor, que ofrece una interfaz no bloqueante perfecta para la arquitectura orientada al rendimiento de FastAPI y FastHTML.

Por qué este stack destaca

Combinar estas tecnologías crea un entorno de desarrollo especialmente productivo:

• Tipado con Pydantic y MongoDB: FastHTML aprovecha Pydantic para el modelado y la validación de datos. Estos modelos se ajustan de forma natural a la estructura documental de MongoDB y ofrecen una experiencia "code-first" que elimina el boilerplate pesado de los ORM.
• Rendimiento asíncrono extremo a extremo: al combinar Motor con el núcleo asíncrono de FastHTML, las operaciones con la base de datos no bloquean el bucle de eventos. Esto garantiza alta concurrencia y baja latencia, cruciales para aplicaciones reactivas y en tiempo real.
• Menos cambios de contexto: puedes gestionar el esquema de la base de datos, la lógica de backend y los componentes de frontend en un ecosistema unificado de Python, acelerando notablemente la entrega.

Configuración y conexión

Hay algunos requisitos previos que necesitas para seguir este tutorial: 

• Python 3.8+
• Instancia de MongoDB: una instalación local o un clúster de MongoDB Atlas (recomendado para funciones en tiempo real)
• Conocimientos básicos: familiaridad con decoradores de Python y la estructura básica de HTML.

Inicialización del proyecto

Para mostrar la eficiencia de este stack, implementaremos toda la aplicación, incluida la configuración de la base de datos, los modelos de datos y las rutas reactivas, en un único archivo app.py.

Instalación 

Necesitamos el framework FastHTML, Motor (el driver asíncrono oficial de MongoDB) y Uvicorn como servidor ASGI.

Configuración de la conexión a MongoDB

Usamos AsyncIOMotorClient para establecer una conexión no bloqueante. Así, mientras tu aplicación espera la E/S de la base de datos, puede seguir atendiendo otras solicitudes 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()

Definición del modelo de datos 

En un flujo de trabajo orientado a documentos, el esquema vive en tu código. Usamos Pydantic v2 para salvar la distancia entre el ObjectId BSON de MongoDB y las cadenas de texto estándar de Python. Así garantizamos que cada documento que entra o sale de la base de datos se valida según nuestros requisitos.

Definimos una clase personalizada PyObjectId. Es necesaria porque Pydantic no sabe manejar de forma nativa el tipo ObjectId de MongoDB. Con __get_pydantic_core_schema__ indicamos a Pydantic que trate este tipo como una cadena en JSON pero lo valide como objeto BSON para la base de datos.

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
    )

Diseño y ruta inicial 

El componente de layout es una función de orden superior reutilizable que envuelve nuestras vistas. Así garantizamos que dependencias esenciales como Tailwind CSS para estilos y HTMX para interactividad sean consistentes en toda la aplicación.

Al usar los componentes en mayúscula de FastHTML (como Main y Div), mantenemos una estructura pythónica y limpia que refleja el árbol 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)
    )

Ahora, si ejecutas la app, la interfaz se verá así:

Imagen: captura tras la configuración inicial

Aún no hay tareas. En la siguiente sección añadiremos y actualizaremos tareas.

Implementación CRUD completa

Read: mostrar tareas (GET /)

Para mostrar las tareas, obtenemos todos los documentos y los renderizamos en un componente. Sin embargo, en un caso real, la base de datos podría estar fuera de línea. Nuestra versión de TaskList gestiona esto con elegancia ofreciendo pasos de diagnóstico directamente en la interfaz.

Obtención de datos resiliente 

Usamos collection.find().to_list(length=None) para recuperar documentos de forma asíncrona. Al envolverlo en un bloque try/except, podemos detectar si MongoDB está desconectado y ofrecer comentarios inmediatos al usuario.

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: añadir una nueva tarea

El flujo de creación usa un formulario HTML con atributos de HTMX. En la versión actualizada, extraemos explícitamente los datos del formulario desde el objeto Request y usamos model_dump para preparar el documento antes de insertarlo en 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 ejecutamos el siguiente comando en la 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"

Verás que se ha añadido la nueva tarea:

Imagen: captura tras añadir una tarea

Update: alternar el estado de finalización

Para gestionar actualizaciones, usamos una solicitud PATCH. Esto demuestra un "micro-refresh": solo se actualiza en la base de datos la fila concreta de la tarea y se vuelve a renderizar en la UI usando su ID específico.

@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: eliminar una tarea

La eliminación se inicia desde HTMX. Una vez que MongoDB confirma el borrado, el servidor devuelve una respuesta Empty(). HTMX interpreta esta respuesta vacía como una señal para eliminar el elemento objetivo (closest div) del 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 hacemos…

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

…la tarea se elimina.

Imagen: captura tras eliminar una tarea 

Puedes ver el script completo en GitHub.

Conclusión

Aprovechando un stack moderno y centrado en Python, has construido una aplicación reactiva que evita la complejidad tradicional de los frameworks de JavaScript en el cliente. Esta arquitectura ofrece varias ventajas clave para el desarrollo web actual. La sinergia entre la UI basada en componentes de FastHTML y el modelo documental flexible de MongoDB te permite mantener la lógica de negocio, la integridad de los datos y la presentación en un único ecosistema cohesionado. Este enfoque "todo en Python" reduce de forma significativa la carga de desarrollo y la complejidad del despliegue.

Próximos pasos para lectores

• Autenticación de usuarios: usa dependencias de FastAPI para restringir listas de tareas a usuarios concretos, guardando user_id en el documento de la tarea.
• Consultas avanzadas: usa el framework de agregación de MongoDB o filtros simples para añadir vistas de "Activas" y "Completadas" a tu interfaz.
• Despliegue: despliega tu app.py con Uvicorn detrás de un proxy inverso NGINX para un rendimiento de nivel producción.