FastHTML और MongoDB के साथ रीयल-टाइम टास्क मैनेजर बनाएँ

FastHTML और MongoDB आधुनिक वेब विकास के लिए एक तेज़, Python-नेटिव तरीका प्रदान करते हैं। इस ट्यूटोरियल में, हम एक रिएक्टिव, रीयल-टाइम टास्क मैनेजर एप्लिकेशन बनाएंगे, जो एक ही, आसान-से-मेंटेन Python फ़ाइल में पूर्ण CRUD (Create, Read, Update, Delete) लाइफ़साइकल प्रदर्शित करेगा। Python में MongoDB का हाथों-हाथ अनुभव लेने के लिए, मैं कोर्स Introduction to MongoDB in Python की सलाह देता/देती हूँ।

FastHTML क्या है?

FastHTML एक मिनिमलिस्ट, उच्च-प्रदर्शन फ़्रेमवर्क है जो FastAPI की नींव पर बनाया गया है। यह एक Pythonic HTML पैरेडाइम प्रस्तुत करता है, जिससे डेवलपर पारंपरिक टेम्पलेट्स की बजाय पुन: प्रयोज्य Python फ़ंक्शंस का उपयोग करके पूरा फ्रंटएंड बना सकते हैं।

इसकी मूल ताकत HTMX के साथ स्वदेशी इंटीग्रेशन में निहित है। साधारण HTML एट्रिब्यूट्स का उपयोग करके सर्वर-साइड अपडेट करवाते हुए, HTMX आपको बिना भारी JavaScript बिल्ड स्टैक की जटिलता के डायनेमिक, सिंगल-पेज एप्लिकेशन अनुभव बनाने देता है।

MongoDB क्या है?

MongoDB अग्रणी जनरल-पर्पज़, डॉक्युमेंट-आधारित NoSQL डेटाबेस है। इसका लचीला स्कीमा और JSON-जैसे BSON डॉक्युमेंट्स का उपयोग इसे आधुनिक, आवर्तक (iterative) विकास के लिए आदर्श बनाता है।

Python के लिए, हम आधिकारिक असिंक्रोनस ड्राइवर Motor का उपयोग करते हैं, जो FastAPI और FastHTML की परफ़ॉर्मेंस-उन्मुख आर्किटेक्चर के लिए एक नॉन-ब्लॉकिंग इंटरफ़ेस प्रदान करता है।

यह स्टैक क्यों उत्कृष्ट है

इन तकनीकों का संयोजन एक अत्यंत उत्पादक विकास वातावरण बनाता है:

• Pydantic और MongoDB टाइप सेफ़्टी: FastHTML डेटा मॉडलिंग और वैलिडेशन के लिए Pydantic का लाभ उठाता है। ये मॉडल स्वाभाविक रूप से MongoDB के डॉक्युमेंट स्ट्रक्चर पर मैप होते हैं, जिससे एक "कोड-फ़र्स्ट" अनुभव मिलता है जो भारी ऑब्जेक्ट-रिलेशनल मैपिंग (ORM) बोइलरप्लेट की आवश्यकता समाप्त करता है।
• एंड-टू-एंड असिंक परफ़ॉर्मेंस: Motor को FastHTML के असिंक्रोनस कोर के साथ पेयर करके, डेटाबेस ऑपरेशन्स कभी भी इवेंट लूप को ब्लॉक नहीं करते। यह उच्च कंकरेन्सी और कम लेटेंसी सुनिश्चित करता है, जो रीयल-टाइम, रिएक्टिव एप्लिकेशन्स के लिए आवश्यक है।
• कॉन्टेक्स्ट स्विचिंग में कमी: डेवलपर्स एकीकृत Python इकोसिस्टम के भीतर डेटाबेस स्कीमा, बैकएंड लॉजिक और फ्रंटएंड कम्पोनेंट्स को मैनेज कर सकते हैं, जिससे डिलिवरी स्पीड में उल्लेखनीय वृद्धि होती है।

सेटअप और कनेक्शन

इस ट्यूटोरियल के साथ चलते हुए आपको कुछ पूर्वापेक्षाएँ चाहिए होंगी: 

• Python 3.8+
• MongoDB इंस्टेंस: लोकल इंस्टॉलेशन या MongoDB Atlas क्लस्टर (रीयल-टाइम फ़ीचर्स के लिए अनुशंसित)
• बुनियादी जानकारी: Python डेकोरेटर और बेसिक HTML संरचना से परिचित होना।

प्रोजेक्ट इनिशियलाइज़ेशन

इस स्टैक की दक्षता दिखाने के लिए, हम पूरे एप्लिकेशन—डेटाबेस कॉन्फ़िगरेशन, डेटा मॉडल्स, और रिएक्टिव रूट्स सहित—एक ही app.py फ़ाइल में लागू करेंगे।

इंस्टॉलेशन 

हमें FastHTML फ़्रेमवर्क, Motor (आधिकारिक असिंक्रोनस MongoDB ड्राइवर) और ASGI सर्वर के लिए Uvicorn की आवश्यकता होगी।

MongoDB कनेक्शन सेटअप

हम AsyncIOMotorClient का उपयोग नॉन-ब्लॉकिंग कनेक्शन स्थापित करने के लिए करते हैं। इससे सुनिश्चित होता है कि जब आपका एप्लिकेशन डेटाबेस 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()

डेटा मॉडल परिभाषित करना 

डॉक्युमेंट-ओरिएंटेड वर्कफ़्लो में, स्कीमा आपके एप्लिकेशन कोड में होता है। हम Pydantic v2 का उपयोग MongoDB के BSON ObjectId और मानक Python स्ट्रिंग्स के बीच पुल बनाने के लिए करते हैं। इससे सुनिश्चित होता है कि हमारे डेटाबेस में जाने या उससे आने वाला हर डॉक्युमेंट हमारी आवश्यकताओं के अनुसार वैलिडेट हो।

हम एक कस्टम PyObjectId क्लास परिभाषित करते हैं। यह आवश्यक है क्योंकि Pydantic स्वाभाविक रूप से MongoDB के ObjectId टाइप को संभालना नहीं जानता। __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 ऑब्जेक्ट से फ़ॉर्म डेटा निकालते हैं और MongoDB में इंसर्शन के लिए डॉक्युमेंट तैयार करने हेतु model_dump का उपयोग करते हैं।

@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 अनुरोध का उपयोग करते हैं। यह एक "माइक्रो-रिफ्रेश" दिखाता है, जहाँ केवल एक टास्क रो को डेटाबेस में अपडेट किया जाता है और उसके विशिष्ट ID का उपयोग करके UI में फिर से रेंडर किया जाता है।

@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 इस खाली प्रतिक्रिया को DOM से लक्ष्य एलिमेंट (निकटतम div) हटाने के संकेत के रूप में इंटरप्रेट करता है।

@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-केंद्रित स्टैक का लाभ उठाकर, आपने एक रिएक्टिव एप्लिकेशन बनाया है जो क्लाइंट-साइड JavaScript फ़्रेमवर्क्स की पारंपरिक जटिलता से बचता है। यह विशेष आर्किटेक्चर आधुनिक वेब विकास के लिए कई प्रमुख लाभ प्रदान करता है। FastHTML के कंपोनेंट-आधारित UI और MongoDB के लचीले डॉक्युमेंट मॉडल के बीच की समन्वयता आपको एक ही, सुसंगत इकोसिस्टम में बिज़नेस लॉजिक, डेटा इंटेग्रिटी और प्रेजेंटेशन बनाए रखने देती है। यह "ऑल-इन-Python" तरीका विकास ओवरहेड और डिप्लॉयमेंट जटिलता को उल्लेखनीय रूप से कम करता है।

पाठकों के लिए अगले कदम

• उपयोगकर्ता प्रमाणीकरण: FastAPI डिपेंडेंसीज़ का उपयोग करके टास्क सूचियों को विशिष्ट उपयोगकर्ताओं तक सीमित करें, और टास्क डॉक्युमेंट में user_id स्टोर करें।
• एडवांस्ड क्वेरीज़: MongoDB का एग्रीगेशन फ़्रेमवर्क या सरल फ़िल्टर्स का उपयोग करके अपनी UI में "Active" और "Completed" व्यूज़ जोड़ें।
• डिप्लॉयमेंट: अपने app.py को प्रोडक्शन-ग्रेड परफ़ॉर्मेंस के लिए NGINX रिवर्स प्रॉक्सी के पीछे Uvicorn के साथ डिप्लॉय करें।