Zbuduj menedżer zadań w czasie rzeczywistym z FastHTML i MongoDB

FastHTML i MongoDB oferują szybkie, natywne dla Pythona podejście do nowoczesnego tworzenia aplikacji webowych. W tym tutorialu zbudujemy reaktywną aplikację do zarządzania zadaniami w czasie rzeczywistym, pokazując kompletny cykl CRUD (Create, Read, Update, Delete) w jednym, łatwym w utrzymaniu pliku Pythona. Jeśli chcesz poćwiczyć korzystanie z MongoDB w Pythonie, polecam kurs Introduction to MongoDB in Python.

Czym jest FastHTML?

FastHTML to minimalistyczny, wydajny framework zbudowany na fundamentach FastAPI. Wprowadza „pythoniczne” podejście do HTML, dzięki któremu możesz tworzyć całe frontendowe warstwy za pomocą wielokrotnego użytku funkcji Pythona zamiast tradycyjnych szablonów.

Jego główna siła to natywna integracja z HTMX. Dzięki prostym atrybutom HTML, które wywołują aktualizacje po stronie serwera, HTMX pozwala tworzyć dynamiczne doświadczenia typu single-page bez złożoności ciężkiego stosu JavaScript.

Czym jest MongoDB?

MongoDB to wiodąca, ogólnego przeznaczenia, dokumentowa baza NoSQL. Elastyczny schemat i wykorzystanie dokumentów BSON przypominających JSON sprawiają, że świetnie nadaje się do nowoczesnego, iteracyjnego rozwoju.

W Pythonie używamy oficjalnego asynchronicznego sterownika Motor, który zapewnia nieblokujący interfejs idealnie dopasowany do architektury nastawionej na wydajność FastAPI i FastHTML.

Dlaczego ten stos się wyróżnia

Połączenie tych technologii tworzy wyjątkowo produktywne środowisko programistyczne:

• Bezpieczeństwo typów Pydantic i MongoDB: FastHTML wykorzystuje Pydantic do modelowania i walidacji danych. Te modele naturalnie mapują się na strukturę dokumentów MongoDB, zapewniając „code-first” doświadczenie i eliminując potrzebę rozbudowanego boilerplate’u ORM.
• Asynchroniczna wydajność end-to-end: Łącząc Motor z asynchronicznym rdzeniem FastHTML, operacje na bazie nigdy nie blokują pętli zdarzeń. Zapewnia to wysoką współbieżność i niskie opóźnienia, kluczowe dla aplikacji czasu rzeczywistego.
• Mniej przełączania kontekstu: Programiści mogą zarządzać schematem bazy, logiką backendu i komponentami frontendu w jednym ekosystemie Pythona, znacznie przyspieszając dostarczanie.

Konfiguracja i połączenie

Jest kilka rzeczy, które musisz przygotować, aby śledzić ten tutorial: 

• Python 3.8+
• Instancja MongoDB: lokalna instalacja lub klaster MongoDB Atlas (zalecane dla funkcji czasu rzeczywistego)
• Podstawowa wiedza: znajomość dekoratorów Pythona i podstawowej struktury HTML.

Inicjalizacja projektu

Aby pokazać skuteczność tego stosu, zaimplementujemy całą aplikację, w tym konfigurację bazy, modele danych i reaktywne trasy, w jednym pliku app.py.

Instalacja 

Potrzebujemy frameworka FastHTML, Motor (oficjalny asynchroniczny sterownik MongoDB) oraz Uvicorn jako serwera ASGI.

Konfiguracja połączenia z MongoDB

Używamy AsyncIOMotorClient, aby ustanowić nieblokujące połączenie. Dzięki temu, gdy aplikacja czeka na operacje I/O bazy danych, może równolegle obsługiwać inne żądania.

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

Definiowanie modelu danych 

W podejściu dokumentowym schemat żyje w kodzie aplikacji. Używamy Pydantic v2, aby zbudować most między ObjectId BSON z MongoDB a standardowymi łańcuchami znaków Pythona. Zapewnia to, że każdy dokument trafiający do bazy lub z niej wychodzący jest walidowany względem naszych wymagań.

Definiujemy własną klasę PyObjectId. Jest to konieczne, ponieważ Pydantic nie obsługuje natywnie typu ObjectId z MongoDB. Dzięki __get_pydantic_core_schema__ wskazujemy Pydantic, by traktował ten typ jako string w JSON, ale walidował go jako obiekt BSON dla bazy.

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
    )

Układ i początkowa trasa 

Komponent układu to wielokrotnego użytku funkcja wyższego rzędu, która opakowuje nasze widoki. Gwarantuje to, że kluczowe zależności, jak Tailwind CSS do stylowania i HTMX do interaktywności, są spójne w całej aplikacji.

Używając komponentów FastHTML z wielkiej litery (takich jak Main i Div), utrzymujemy czystą, pythoniczną strukturę, która odzwierciedla drzewo 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)
    )

Po uruchomieniu aplikacji interfejs będzie wyglądał tak:

Obraz: zrzut ekranu po konfiguracji

Na razie nie ma żadnych zadań. W kolejnej sekcji dodamy i zaktualizujemy zadania.

Pełna implementacja CRUD

Read: Wyświetlanie zadań (GET /)

Aby wyświetlić zadania, pobieramy wszystkie dokumenty i renderujemy je w komponencie. W rzeczywistości baza może być jednak offline. Nasz zaktualizowany TaskList obsługuje to elegancko, podając wskazówki rozwiązywania problemów bezpośrednio w UI.

Odporne pobieranie danych 

Używamy collection.find().to_list(length=None) do asynchronicznego pobierania dokumentów. Opakowując to w blok try/except, możemy wykryć rozłączenie z MongoDB i natychmiast poinformować użytkownika.

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: Dodawanie nowego zadania

Tworzenie wykorzystuje formularz HTML wzbogacony o atrybuty HTMX. W zaktualizowanej wersji jawnie wyciągamy dane formularza z obiektu Request i używamy model_dump do przygotowania dokumentu do wstawienia w 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')

Jeśli uruchomisz poniższy kod w terminalu:

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"

Zobaczysz, że nowe zadanie zostało dodane:

Obraz: zrzut ekranu po dodaniu zadania

Update: Przełączanie statusu ukończenia

Aby obsłużyć aktualizacje, używamy żądania PATCH. Pokazuje to „mikroodświeżenie”, w którym tylko pojedynczy wiersz zadania jest zaktualizowany w bazie i ponownie wyrenderowany w UI na podstawie jego konkretnego 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: Usuwanie zadania

Usuwanie inicjuje HTMX. Gdy MongoDB potwierdzi usunięcie, serwer zwraca odpowiedź Empty(). HTMX interpretuje tę pustą odpowiedź jako sygnał do usunięcia docelowego elementu (najbliższego diva) z 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()

Jeśli wykonamy…

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

…zadanie zostanie usunięte.

Obraz: zrzut ekranu po usunięciu 

Cały skrypt znajdziesz na GitHubie.

Podsumowanie

Wykorzystując nowoczesny, zorientowany na Pythona stos technologiczny, zbudowałeś reaktywną aplikację bez typowej złożoności frameworków JavaScript po stronie klienta. Ta konkretna architektura zapewnia kilka kluczowych zalet dla współczesnego web developmentu. Synergia między komponentowym UI FastHTML a elastycznym modelem dokumentowym MongoDB pozwala utrzymać logikę biznesową, integralność danych i warstwę prezentacji w jednym, spójnym ekosystemie. To „wszystko w Pythonie” znacząco redukuje narzut deweloperski i złożoność wdrożeń.

Kolejne kroki dla czytelników

• Uwierzytelnianie użytkowników: użyj zależności FastAPI, aby ograniczyć listy zadań do konkretnych użytkowników, zapisując user_id w dokumencie zadania.
• Zaawansowane zapytania: użyj frameworka agregacji MongoDB lub prostych filtrów, aby dodać widoki „Aktywne” i „Ukończone” w UI.
• Wdrożenie: wdroż app.py z Uvicornem za NGINX jako reverse proxy, aby uzyskać wydajność produkcyjną.