สร้างตัวจัดการงานแบบเรียลไทม์ด้วย FastHTML และ MongoDB

FastHTML และ MongoDB มอบแนวทางการพัฒนาเว็บสมัยใหม่ที่รวดเร็วและเนทีฟสำหรับ Python ในบทความนี้ เราจะสร้างแอปตัวจัดการงานแบบ reactive และเรียลไทม์ โดยสาธิตวงจร CRUD (Create, Read, Update, Delete) แบบครบถ้วนภายในไฟล์ Python เดียวที่ดูแลง่าย หากต้องการทดลองใช้ MongoDB กับ Python แบบลงมือทำ ขอแนะนำคอร์ส Introduction to MongoDB in Python।

FastHTML คืออะไร?

FastHTML คือเฟรมเวิร์กมินิมัล ประสิทธิภาพสูง ที่สร้างบนรากฐานของ FastAPI มันนำเสนอกระบวนทัศน์ HTML แบบ Pythonic ทำให้นักพัฒนาสามารถสร้างฟรอนต์เอนด์ทั้งหมดด้วยฟังก์ชัน Python ที่นำกลับมาใช้ซ้ำได้ แทนการใช้เทมเพลตแบบเดิม

จุดแข็งหลักอยู่ที่การผสานรวมกับ HTMX แบบเนทีฟ โดยใช้แอตทริบิวต์ HTML ที่เรียบง่ายเพื่อขับเคลื่อนการอัปเดตฝั่งเซิร์ฟเวอร์ HTMX ช่วยให้สร้างประสบการณ์แบบ single-page application แบบไดนามิกได้ โดยไม่ต้องพึ่งสแต็กที่หนักด้วย JavaScript

MongoDB คืออะไร?

MongoDB เป็นฐานข้อมูล NoSQL ที่อิงเอกสารแบบ general-purpose ชั้นนำ สคีมาที่ยืดหยุ่นและการใช้เอกสาร BSON ที่คล้าย JSON ทำให้เหมาะกับการพัฒนาสมัยใหม่ที่วนซ้ำได้รวดเร็ว

สำหรับ Python เราใช้ไดรเวอร์แบบอะซิงโครนัสอย่างเป็นทางการชื่อ Motor ซึ่งให้ส่วนติดต่อแบบ non-blocking ที่เหมาะอย่างยิ่งกับสถาปัตยกรรมที่เน้นประสิทธิภาพของ FastAPI และ FastHTML

ทำไมสแต็กนี้จึงโดดเด่น

การผสานเทคโนโลยีเหล่านี้สร้างสภาพแวดล้อมการพัฒนาที่มีประสิทธิผลอย่างยิ่ง:

• ความปลอดภัยของชนิดข้อมูลด้วย Pydantic และ MongoDB: FastHTML ใช้ Pydantic สำหรับการจำลองข้อมูลและการตรวจสอบความถูกต้อง โมเดลเหล่านี้แมปเข้ากับโครงสร้างเอกสารของ MongoDB ได้อย่างเป็นธรรมชาติ มอบประสบการณ์แบบ "เริ่มจากโค้ด" ที่ตัดความจำเป็นของโค้ดส่วนเกินจาก ORM หนัก ๆ
• ประสิทธิภาพอะซิงก์ตั้งแต่ต้นจนจบ: เมื่อจับคู่ Motor กับแกนอะซิงก์ของ FastHTML การทำงานกับฐานข้อมูลจะไม่บล็อก event loop รับรองความพร้อมใช้งานพร้อมกันสูงและความหน่วงต่ำ ซึ่งสำคัญสำหรับแอปแบบเรียลไทม์และ reactive
• ลดการสลับบริบท: นักพัฒนาสามารถจัดการสคีมาฐานข้อมูล ลอจิกฝั่งแบ็กเอนด์ และคอมโพเนนต์ฟรอนต์เอนด์ภายในระบบนิเวศของ Python ที่เป็นหนึ่งเดียว เพิ่มความเร็วในการส่งมอบงานอย่างมาก

การตั้งค่าและการเชื่อมต่อ

มีข้อกำหนดเบื้องต้นบางอย่างที่ต้องมีเพื่อทำตามบทเรียนนี้: 

• Python 3.8+
• อินสแตนซ์ MongoDB: การติดตั้งแบบโลคอลหรือ MongoDB Atlas คลัสเตอร์ (แนะนำสำหรับฟีเจอร์เรียลไทม์)
• ความรู้พื้นฐาน: คุ้นเคยกับตัวตกแต่งของ Python และโครงสร้าง HTML เบื้องต้น

การเริ่มต้นโปรเจกต์

เพื่อแสดงประสิทธิภาพของสแต็กนี้ เราจะทำให้ทั้งแอปเสร็จภายในไฟล์เดียว app.py รวมถึงการตั้งค่าฐานข้อมูล โมเดลข้อมูล และเส้นทางแบบ reactive

การติดตั้ง 

เราต้องใช้เฟรมเวิร์ก FastHTML, Motor (ไดรเวอร์ MongoDB แบบอะซิงก์อย่างเป็นทางการ) และ Uvicorn สำหรับ ASGI server

การตั้งค่าการเชื่อมต่อ MongoDB

เราใช้ AsyncIOMotorClient เพื่อสร้างการเชื่อมต่อแบบ non-blocking ทำให้ในขณะที่แอปกำลังรอ I/O ของฐานข้อมูล ก็ยังประมวลผลคำขออื่น ๆ ที่เกิดขึ้นพร้อมกันต่อไปได้

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()ขับเคลื่อนโดย

การกำหนดโมเดลข้อมูล 

ในเวิร์กโฟลว์แบบ document-oriented สคีมาอยู่ในโค้ดแอปพลิเคชันของคุณ เราใช้ Pydantic v2 เพื่อเชื่อมช่องว่างระหว่าง BSON ObjectId ของ MongoDB และสตริงมาตรฐานของ Python เพื่อให้แน่ใจว่าเอกสารทุกฉบับที่เข้าออกฐานข้อมูลจะถูกตรวจสอบตามข้อกำหนดของเรา

เราสร้างคลาสกำหนดเองชื่อ PyObjectId ซึ่งจำเป็นเพราะ Pydantic ไม่รู้วิธีจัดการชนิด ObjectId ของ MongoDB โดยกำเนิด ด้วยการใช้ __get_pydantic_core_schema__ เราบอกให้ Pydantic ปฏิบัติต่อชนิดนี้เป็นสตริงใน JSON แต่ตรวจสอบความถูกต้องเป็นวัตถุ BSON สำหรับฐานข้อมูล

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
    )ขับเคลื่อนโดย

เลเอาต์และเส้นทางเริ่มต้น 

คอมโพเนนต์เลเอาต์เป็นฟังก์ชันระดับสูงที่นำกลับมาใช้ซ้ำได้ ใช้ห่อมุมมองของเรา เพื่อให้แน่ใจว่าการพึ่งพาที่สำคัญ เช่น Tailwind CSS สำหรับการตกแต่ง และ HTMX สำหรับความโต้ตอบ มีความสอดคล้องกันทั่วทั้งแอป

ด้วยการใช้คอมโพเนนต์ตัวพิมพ์ใหญ่ของ FastHTML (อย่างเช่น Main และ Div) เรารักษาโครงสร้างแบบ Pythonic ที่สะอาดตา ซึ่งสะท้อนโครงสร้างต้นไม้ของ 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)
    )ขับเคลื่อนโดย

ตอนนี้ หากรันแอป UI จะมีหน้าตาแบบนี้:

ภาพ: สกรีนช็อตหลังตั้งค่าเสร็จ

ยังไม่มียานใด ๆ เราจะเพิ่มและอัปเดตงานในส่วนถัดไป

การทำ CRUD แบบครบถ้วน

Read: แสดงงาน (GET /)

เพื่อแสดงงาน เราดึงเอกสารทั้งหมดและเรนเดอร์ภายในคอมโพเนนต์ อย่างไรก็ตาม ในโลกจริง ฐานข้อมูลอาจออฟไลน์ เวอร์ชันอัปเดตของ TaskList จะจัดการอย่างนุ่มนวล โดยแสดงขั้นตอนการแก้ปัญหาโดยตรงใน UI

การดึงข้อมูลที่ทนทาน 

เราใช้ collection.find().to_list(length=None) เพื่อดึงเอกสารแบบอะซิงก์ ด้วยการห่อในบล็อก try/except เราสามารถตรวจจับได้หาก MongoDB ตัดการเชื่อมต่อ และให้ข้อเสนอแนะกับผู้ใช้ได้ทันที

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: การเพิ่มงานใหม่

โฟลว์การสร้างใช้ฟอร์ม HTML ที่เสริมด้วยแอตทริบิวต์ HTMX ในเวอร์ชันอัปเดตนี้ เราดึงข้อมูลฟอร์มจากออบเจ็กต์ Request โดยตรง และใช้ model_dump เพื่อเตรียมเอกสารสำหรับแทรกลง 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')ขับเคลื่อนโดย

หากรันโค้ดด้านล่างในเทอร์มินัล:

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"ขับเคลื่อนโดย

จะเห็นว่างานใหม่นี้ถูกเพิ่มแล้ว:

ภาพ: สกรีนช็อตหลังเพิ่มงาน

Update: สลับสถานะเสร็จสิ้น

สำหรับการอัปเดต เราใช้คำขอ PATCH ซึ่งสาธิต "micro-refresh" ที่อัปเดตเฉพาะแถวของงานในฐานข้อมูลและเรนเดอร์ใหม่ใน UI โดยอ้างอิงตาม 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: ลบงาน

การลบถูกเริ่มต้นโดย HTMX เมื่อ MongoDB ยืนยันการลบ เซิร์ฟเวอร์จะส่งการตอบกลับแบบ Empty() HTMX จะตีความการตอบกลับว่างนี้เป็นสัญญาณให้ลบองค์ประกอบเป้าหมาย (div ที่อยู่ใกล้ที่สุด) ออกจาก 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()ขับเคลื่อนโดย

หากเราทำ…

curl -X DELETE http://localhost:8000/delete-task/695968244236010c04f313faขับเคลื่อนโดย

…งานนั้นจะถูกลบ

ภาพ: สกรีนช็อตหลังการลบ 

ดูสคริปต์ทั้งหมดได้ที่ GitHub.

สรุป

ด้วยการใช้สแต็กสมัยใหม่ที่เน้น Python คุณได้สร้างแอปแบบ reactive ที่หลีกเลี่ยงความซับซ้อนแบบดั้งเดิมของเฟรมเวิร์ก JavaScript ฝั่งไคลเอนต์ สถาปัตยกรรมนี้มอบข้อได้เปรียบสำคัญหลายประการสำหรับการพัฒนาเว็บสมัยใหม่ การทำงานร่วมกันระหว่าง UI แบบคอมโพเนนต์ของ FastHTML และโมเดลเอกสารที่ยืดหยุ่นของ MongoDB ทำให้คงไว้ซึ่งลอจิกธุรกิจ ความถูกต้องของข้อมูล และการนำเสนอภายในระบบนิเวศเดียวที่เหนียวแน่น แนวทางแบบ "ทั้งหมดใน Python" ช่วยลดภาระการพัฒนาและความซับซ้อนในการดีพลอยอย่างมีนัยสำคัญ

ขั้นถัดไปสำหรับผู้อ่าน

• การยืนยันตัวตนผู้ใช้: ใช้ dependencies ของ FastAPI เพื่อจำกัดรายการงานให้เฉพาะผู้ใช้ เก็บ user_id ไว้ในเอกสารงาน
• คิวรีขั้นสูง: ใช้ เฟรมเวิร์ก aggregation ของ MongoDB หรือฟิลเตอร์ง่าย ๆ เพื่อเพิ่มมุมมอง "Active" และ "Completed" ให้กับ UI
• การดีพลอย: ดีพลอย app.py ของคุณด้วย Uvicorn หลังพร็อกซีแบบ reverse ของ NGINX เพื่อประสิทธิภาพระดับโปรดักชัน