Создайте менеджер задач в реальном времени на FastHTML и MongoDB

FastHTML и MongoDB предлагают высокоскоростной, нативный для Python подход к современной веб-разработке. В этом руководстве мы создадим реактивное приложение для управления задачами в реальном времени, продемонстрировав полный жизненный цикл CRUD (создание, чтение, обновление, удаление) в одном поддерживаемом файле на Python. Чтобы получить практический опыт работы с MongoDB в Python, рекомендую курс Introduction to MongoDB in Python.

Что такое FastHTML?

FastHTML — это минималистичный высокопроизводительный фреймворк, построенный на основе FastAPI. Он вводит питонический подход к HTML, позволяя создавать целые фронтенды с помощью многократно используемых функций на Python вместо традиционных шаблонов.

Его ключевая сила — нативная интеграция с HTMX. Используя простые HTML-атрибуты для инициирования серверных обновлений, HTMX позволяет создавать динамичные впечатления уровня SPA без усложнения JavaScript‑стеком сборки.

Что такое MongoDB?

MongoDB — ведущая универсальная документо-ориентированная NoSQL-база данных. Гибкая схема и использование BSON-документов, похожих на JSON, делают её идеальной для современной итеративной разработки.

Для Python мы используем официальный асинхронный драйвер Motor, предоставляющий неблокирующий интерфейс, идеально подходящий к производительной архитектуре FastAPI и FastHTML.

Почему этот стек силён

Сочетание этих технологий создаёт исключительно продуктивную среду разработки:

• Безопасность типов Pydantic и MongoDB: FastHTML использует Pydantic для моделирования и валидации данных. Эти модели естественно отображаются на структуру документов MongoDB, обеспечивая «code-first»-подход и устраняя необходимость в громоздком шаблонном коде ORM.
• Сквозная асинхронная производительность: Благодаря паре Motor и асинхронному ядру FastHTML операции с базой данных не блокируют цикл событий. Это обеспечивает высокую конкурентность и низкую задержку — критично для реактивных приложений в реальном времени.
• Меньше переключений контекста: Разработчики управляют схемой БД, бэкенд-логикой и фронтенд-компонентами в единой экосистеме Python, значительно ускоряя поставку.

Настройка и подключение

Есть несколько предварительных условий, чтобы следовать этому руководству: 

• Python 3.8+
• Экземпляр MongoDB: локальная установка или кластер MongoDB Atlas (рекомендуется для функций реального времени)
• Базовые знания: знакомство с декораторами Python и базовой структурой HTML.

Инициализация проекта

Чтобы продемонстрировать эффективность стека, мы реализуем всё приложение — настройку базы данных, модели данных и реактивные маршруты — в одном файле app.py.

Установка 

Нам понадобятся фреймворк FastHTML, Motor (официальный асинхронный драйвер MongoDB) и Uvicorn для ASGI-сервера.

Настройка подключения к MongoDB

Мы используем AsyncIOMotorClient для установления неблокирующего соединения. Это гарантирует, что во время ожидания операций ввода-вывода базы данных приложение продолжит обрабатывать другие параллельные запросы.

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, чтобы связать 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), мы поддерживаем чистую, питоническую структуру, зеркально отражающую 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)
    )

Теперь, если вы запустите приложение, интерфейс будет выглядеть так:

Изображение: скриншот после настройки

Пока задач нет. В следующем разделе мы добавим и обновим задачи.

Полная реализация CRUD

Read: отображение задач (GET /)

Чтобы отобразить задачи, мы извлекаем все документы и рендерим их внутри компонента. Однако в реальном проекте база может быть недоступна. Наш обновлённый TaskList обрабатывает это корректно, предоставляя подсказки по устранению неполадок прямо в интерфейсе.

Отказоустойчивое получение данных 

Мы используем 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. Это демонстрирует «микрообновление»: в базе меняется только одна строка задачи, а в интерфейсе она перерисовывается по её конкретному 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, вы создали реактивное приложение без традиционной сложности клиентских JavaScript‑фреймворков. Такая архитектура даёт ряд ключевых преимуществ для современной веб-разработки. Синергия между компонентным UI FastHTML и гибкой документной моделью MongoDB позволяет держать бизнес-логику, целостность данных и представление в единой, цельной экосистеме. Этот подход «всё на Python» существенно снижает издержки разработки и сложность развёртывания.

Следующие шаги для читателей

• Аутентификация пользователей: используйте зависимости FastAPI, чтобы ограничить списки задач конкретными пользователями, сохраняя user_id в документе задачи.
• Продвинутые запросы: используйте конвейер агрегирования MongoDB или простые фильтры, чтобы добавить в интерфейс представления «Активные» и «Завершённые».
• Развёртывание: развёртывайте ваш app.py с Uvicorn за обратным прокси NGINX для производительности уровня продакшена.