Bangun Pengelola Tugas Real-Time dengan FastHTML dan MongoDB

FastHTML dan MongoDB menawarkan pendekatan berkecepatan tinggi dan natif Python untuk pengembangan web modern. Dalam tutorial ini, kita akan membangun aplikasi pengelola tugas yang reaktif dan real-time, menampilkan siklus hidup CRUD (Create, Read, Update, Delete) lengkap dalam satu berkas Python yang mudah dikelola. Untuk mendapatkan pengalaman langsung menggunakan MongoDB di Python, saya merekomendasikan kursus Introduction to MongoDB in Python.

Apa itu FastHTML?

FastHTML adalah kerangka kerja minimalis dan berperforma tinggi yang dibangun di atas dasar FastAPI. Ia memperkenalkan paradigma HTML yang Pythonik, memungkinkan pengembang membangun seluruh frontend menggunakan fungsi Python yang dapat digunakan kembali alih-alih templat tradisional.

Kekuatan intinya terletak pada integrasi natif dengan HTMX. Dengan memanfaatkan atribut HTML sederhana untuk menggerakkan pembaruan sisi server, HTMX memungkinkan Anda membuat pengalaman aplikasi satu halaman yang dinamis tanpa kompleksitas tumpukan build yang sarat JavaScript.

Apa itu MongoDB?

MongoDB adalah database NoSQL berbasis dokumen serbaguna terkemuka. Skema yang fleksibel dan penggunaan dokumen BSON mirip JSON menjadikannya ideal untuk pengembangan modern dan iteratif.

Untuk Python, kita menggunakan driver asinkron resmi, Motor, yang menyediakan antarmuka non-blocking yang sangat cocok untuk arsitektur berorientasi performa FastAPI dan FastHTML.

Mengapa Tumpukan Ini Unggul

Menggabungkan teknologi ini menciptakan lingkungan pengembangan yang sangat produktif:

• Keamanan tipe Pydantic dan MongoDB: FastHTML memanfaatkan Pydantic untuk pemodelan dan validasi data. Model-model ini secara alami dipetakan ke struktur dokumen MongoDB, memberikan pengalaman "code-first" yang menghilangkan kebutuhan akan boilerplate object-relational mapping (ORM) yang berat.
• Performa async end-to-end: Dengan memasangkan Motor dengan inti asinkron FastHTML, operasi basis data tidak pernah memblokir event loop. Ini memastikan konkurensi tinggi dan latensi rendah, yang penting untuk aplikasi real-time dan reaktif.
• Pengurangan context switching: Pengembang dapat mengelola skema basis data, logika backend, dan komponen frontend dalam ekosistem Python terpadu, sehingga secara signifikan meningkatkan kecepatan pengiriman.

Penyiapan dan Koneksi

Ada beberapa prasyarat yang perlu Anda penuhi untuk mengikuti tutorial ini: 

• Python 3.8+
• Instans MongoDB: instalasi lokal atau kluster MongoDB Atlas (direkomendasikan untuk fitur real-time)
• Pengetahuan dasar: pemahaman tentang dekorator Python dan struktur HTML dasar.

Inisialisasi proyek

Untuk menunjukkan efisiensi tumpukan ini, kita akan mengimplementasikan seluruh aplikasi, termasuk konfigurasi basis data, model data, dan rute reaktif, dalam satu berkas app.py.

Instalasi 

Kita memerlukan kerangka kerja FastHTML, Motor (driver MongoDB asinkron resmi), dan Uvicorn untuk server ASGI.

Penyiapan koneksi MongoDB

Kita menggunakan AsyncIOMotorClient untuk membangun koneksi non-blocking. Ini memastikan bahwa saat aplikasi Anda menunggu I/O basis data, aplikasi tetap dapat memproses permintaan lain secara konkuren.

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

Mendefinisikan model data 

Dalam alur kerja berorientasi dokumen, skema berada di kode aplikasi Anda. Kita menggunakan Pydantic v2 untuk menjembatani kesenjangan antara BSON ObjectId milik MongoDB dan string Python standar. Ini memastikan setiap dokumen yang masuk atau keluar dari basis data kita divalidasi sesuai kebutuhan.

Kita mendefinisikan kelas kustom PyObjectId. Ini diperlukan karena Pydantic tidak secara native mengetahui cara menangani tipe ObjectId milik MongoDB. Dengan menggunakan __get_pydantic_core_schema__, kita memberi tahu Pydantic untuk memperlakukan tipe ini sebagai string dalam JSON tetapi memvalidasinya sebagai objek BSON untuk basis data.

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
    )

Tata letak dan rute awal 

Komponen tata letak adalah fungsi tingkat tinggi yang dapat digunakan kembali yang membungkus tampilan kita. Ini memastikan dependensi penting, seperti Tailwind CSS untuk styling dan HTMX untuk interaktivitas, konsisten di seluruh aplikasi.

Dengan menggunakan komponen huruf kapital milik FastHTML (seperti Main dan Div), kita menjaga struktur yang bersih dan Pythonik yang mencerminkan pohon 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)
    )

Sekarang, jika Anda menjalankan aplikasi, UI Anda akan terlihat seperti ini:

Gambar: tangkapan layar setelah penyiapan

Belum ada tugas. Kita akan menambahkan dan memperbarui tugas pada bagian berikutnya.

Implementasi CRUD Lengkap

Read: Menampilkan tugas (GET /)

Untuk menampilkan tugas, kita mengambil semua dokumen dan merendernya dalam sebuah komponen. Namun, dalam skenario dunia nyata, basis data mungkin sedang offline. TaskList yang diperbarui ini menanganinya dengan baik dengan memberikan langkah pemecahan masalah langsung di UI.

Pengambilan data yang tangguh 

Kita menggunakan collection.find().to_list(length=None) untuk mengambil dokumen secara asinkron. Dengan membungkusnya dalam blok try/except, kita dapat mendeteksi apakah MongoDB terputus dan memberikan umpan balik langsung kepada pengguna.

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: Menambahkan tugas baru

Alur pembuatan menggunakan formulir HTML yang ditingkatkan dengan atribut HTMX. Dalam versi yang diperbarui, kita secara eksplisit mengekstrak data formulir dari objek Request dan menggunakan model_dump untuk menyiapkan dokumen bagi penyisipan 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')

Jika kita menjalankan kode berikut di 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"

Anda akan melihat bahwa tugas baru ini ditambahkan:

Gambar: tangkapan layar setelah menambahkan tugas

Update: Mengalihkan status penyelesaian

Untuk menangani pembaruan, kita menggunakan permintaan PATCH. Ini menunjukkan "micro-refresh", di mana hanya baris tugas tunggal yang diperbarui di basis data dan dirender ulang di UI menggunakan ID spesifiknya.

@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: Menghapus tugas

Penghapusan dipicu oleh HTMX. Setelah MongoDB mengonfirmasi penghapusan, server mengembalikan respons Empty(). HTMX menafsirkan respons kosong ini sebagai sinyal untuk menghapus elemen target (div terdekat) dari 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()

Jika kita melakukan…

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

…tugas tersebut akan dihapus.

Gambar: tangkapan layar setelah penghapusan 

Anda dapat menemukan seluruh skrip di GitHub.

Kesimpulan

Dengan memanfaatkan tumpukan modern yang berpusat pada Python, Anda telah membangun aplikasi reaktif yang menghindari kompleksitas tradisional kerangka kerja JavaScript sisi klien. Arsitektur spesifik ini memberikan beberapa keunggulan kunci untuk pengembangan web modern. Sinergi antara UI berbasis komponen FastHTML dan model dokumen fleksibel MongoDB memungkinkan Anda mempertahankan logika bisnis, integritas data, dan presentasi dalam satu ekosistem yang kohesif. Pendekatan "serba Python" ini secara signifikan mengurangi beban pengembangan dan kompleksitas deployment.

Langkah selanjutnya untuk pembaca

• Autentikasi pengguna: Gunakan dependensi FastAPI untuk membatasi daftar tugas pada pengguna tertentu, dengan menyimpan user_id di dokumen tugas.
• Kueri lanjutan: Gunakan kerangka kerja agregasi MongoDB atau filter sederhana untuk menambahkan tampilan "Aktif" dan "Selesai" ke UI Anda.
• Deployment: Deploy app.py Anda menggunakan Uvicorn di belakang reverse proxy NGINX untuk performa setara produksi.