Construiește un manager de taskuri în timp real cu FastHTML și MongoDB

FastHTML și MongoDB oferă o abordare rapidă, nativă pentru Python, pentru dezvoltarea web modernă. În acest tutorial, vom construi o aplicație reactivă, în timp real, pentru gestionarea taskurilor, demonstrând un ciclu complet CRUD (Create, Read, Update, Delete) într-un singur fișier Python ușor de întreținut. Pentru a căpăta experiență practică în utilizarea MongoDB în Python, îți recomand cursul Introduction to MongoDB in Python.

Ce este FastHTML?

FastHTML este un framework minimalist, de înaltă performanță, construit pe baza FastAPI. Introduce un paradigm HTML „pythonic”, permițând dezvoltatorilor să creeze întregul frontend folosind funcții Python reutilizabile, în locul șabloanelor tradiționale.

Punctul său forte constă în integrarea nativă cu HTMX. Folosind atribute HTML simple pentru a declanșa actualizări pe server, HTMX îți permite să creezi experiențe tip single-page application fără complexitatea unui stack de build puternic dependent de JavaScript.

Ce este MongoDB?

MongoDB este cea mai populară bază de date NoSQL generalistă, bazată pe documente. Schema sa flexibilă și utilizarea documentelor BSON similare cu JSON o fac ideală pentru dezvoltarea modernă, iterativă.

Pentru Python, folosim driverul asincron oficial, Motor, care oferă o interfață non-blocantă, perfect potrivită pentru arhitectura orientată pe performanță a FastAPI și FastHTML.

De ce această stivă excelează

Combinarea acestor tehnologii creează un mediu de dezvoltare extrem de productiv:

• Siguranță tipuri cu Pydantic și MongoDB: FastHTML folosește Pydantic pentru modelarea și validarea datelor. Aceste modele se mapează natural pe structura de documente a MongoDB, oferind o experiență „code-first” care elimină nevoia unui boilerplate ORM greoi.
• Performanță async cap la cap: Prin combinarea Motor cu nucleul asincron al FastHTML, operațiunile pe bază de date nu blochează niciodată event loop-ul. Asta asigură concurență ridicată și latență scăzută, esențiale pentru aplicațiile reactive, în timp real.
• Reducerea schimbării de context: Poți gestiona schema bazei de date, logica backend și componentele frontend într-un ecosistem unificat Python, crescând semnificativ viteza de livrare.

Setare și conectare

Sunt câteva cerințe preliminare pe care trebuie să le îndeplinești pentru a urma acest tutorial: 

• Python 3.8+
• Instanță MongoDB: o instalare locală sau un cluster MongoDB Atlas (recomandat pentru funcționalități în timp real)
• Cunoștințe de bază: familiaritate cu decoratori în Python și structura HTML de bază.

Inițializarea proiectului

Pentru a evidenția eficiența acestei stive, vom implementa întreaga aplicație, inclusiv configurarea bazei de date, modelele de date și rutele reactive, într-un singur fișier app.py.

Instalare 

Avem nevoie de framework-ul FastHTML, Motor (driverul asincron oficial MongoDB) și Uvicorn pentru serverul ASGI.

Configurarea conexiunii la MongoDB

Folosim AsyncIOMotorClient pentru a stabili o conexiune non-blocantă. Astfel, în timp ce aplicația așteaptă I/O de la baza de date, poate continua să proceseze alte cereri concurente.

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

Definirea modelului de date 

Într-un flux de lucru orientat pe documente, schema trăiește în codul aplicației. Folosim Pydantic v2 pentru a face puntea între ObjectId BSON al MongoDB și șirurile standard Python. Astfel, fiecare document care intră sau iese din baza noastră de date este validat conform cerințelor.

Definim o clasă personalizată PyObjectId. Este necesară deoarece Pydantic nu știe nativ cum să gestioneze tipul ObjectId al MongoDB. Folosind __get_pydantic_core_schema__, îi indicăm lui Pydantic să trateze acest tip ca șir în JSON, dar să-l valideze ca obiect BSON pentru baza de date.

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 și ruta inițială 

Componenta de layout este o funcție de ordin superior, reutilizabilă, care „îmbracă” vederile noastre. Astfel, dependențe esențiale precum Tailwind CSS pentru stilizare și HTMX pentru interactivitate sunt consistente în întreaga aplicație.

Folosind componentele cu inițială mare din FastHTML (precum Main și Div), menținem o structură curată, „pythonică”, care reflectă arborele 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)
    )

Acum, dacă rulezi aplicația, interfața ta va arăta astfel:

Imagine: captură de ecran după configurare

Încă nu există taskuri. Vom lucra la adăugarea și actualizarea taskurilor în secțiunea următoare.

Implementare completă CRUD

Read: Afișarea taskurilor (GET /)

Pentru a afișa taskurile, preluăm toate documentele și le randăm într-o componentă. Totuși, într-un scenariu real, baza de date ar putea fi offline. TaskList actualizat gestionează elegant acest lucru, oferind pași de depanare direct în interfață.

Preluare de date rezilientă 

Folosim collection.find().to_list(length=None) pentru a prelua asincron documentele. Încapsulând asta într-un bloc try/except, putem detecta dacă MongoDB este deconectat și putem oferi utilizatorului feedback imediat.

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: Adăugarea unui task nou

Fluxul de creare folosește un formular HTML îmbunătățit cu atribute HTMX. În versiunea actualizată, extragem explicit datele din formular din obiectul Request și folosim model_dump pentru a pregăti documentul pentru inserarea în 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')

Dacă rulăm comanda de mai jos în 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"

Vei putea vedea că acest task nou este adăugat:

Imagine: captură de ecran după adăugarea unui task

Update: Comutarea stării de finalizare

Pentru a gestiona actualizările, folosim o cerere PATCH. Asta demonstrează un „micro-refresh”, în care doar rândul specific al taskului este actualizat în baza de date și re-redat în UI folosind ID-ul său specific.

@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: Ștergerea unui task

Ștergerea este inițiată de HTMX. După ce MongoDB confirmă ștergerea, serverul returnează un răspuns Empty(). HTMX interpretează acest răspuns gol ca un semnal de a elimina elementul țintă (cel mai apropiat div) din 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()

Dacă facem…

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

…taskul este șters.

Imagine: captură de ecran după ștergere 

Poți găsi întregul script pe GitHub.

Concluzie

Folosind o stivă modernă, centrată pe Python, ai construit o aplicație reactivă care evită complexitatea tradițională a framework-urilor JavaScript pe client. Această arhitectură oferă mai multe avantaje-cheie pentru dezvoltarea web modernă. Sinergia dintre UI-ul bazat pe componente al FastHTML și modelul flexibil de documente al MongoDB îți permite să menții logica de business, integritatea datelor și prezentarea într-un singur ecosistem coerent. Această abordare „totul în Python” reduce semnificativ efortul de dezvoltare și complexitatea de deploy.

Pași următori pentru cititori

• Autentificarea utilizatorilor: Folosește dependențe FastAPI pentru a restricționa listele de taskuri la anumiți utilizatori, stocând user_id în documentul taskului.
• Interogări avansate: Folosește framework-ul de agregare al MongoDB sau filtre simple pentru a adăuga vederi „Active” și „Completed” în UI.
• Deploy: Fă deploy fișierului tău app.py folosind Uvicorn în spatele unui reverse proxy NGINX pentru performanță de producție.