FastHTML と MongoDB でリアルタイムのタスクマネージャーを構築する

Python ネイティブのツールで非同期 CRUD と HTMX のインタラクションを実現する完全チュートリアル。

更新 2026年6月17日 · 5 分読む

FastHTML と MongoDB は、高速で Python ネイティブな最新の Web 開発手法を提供します。本チュートリアルでは、リアクティブでリアルタイムなタスクマネージャーアプリケーションを構築し、単一の保守しやすい Python ファイル内で、完全な CRUD（作成・読み取り・更新・削除）ライフサイクルを実演します。Python での MongoDB の実践を進めるには、コース Introduction to MongoDB in Python をおすすめします。

FastHTML とは？

FastHTML は、FastAPI を土台としたミニマルで高性能なフレームワークです。Python 的な HTML パラダイムを導入し、従来のテンプレートではなく、再利用可能な Python 関数でフロントエンド全体を構築できるようにします。

その中核の強みは、HTMX とのネイティブ統合にあります。サーバーサイドの更新をシンプルな HTML 属性で駆動する HTMX により、JavaScript に大きく依存したビルドスタックの複雑さなしに、動的なシングルページアプリケーション体験を実現できます。

MongoDB とは？

MongoDB は、汎用のドキュメント指向 NoSQL データベースのリーダーです。柔軟なスキーマと JSON に似た BSON ドキュメントを用いることで、現代的で反復的な開発に最適です。

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__ を用いて、この型を JSON では文字列として扱い、データベースでは BSON オブジェクトとして検証するよう Pydantic に指示します。

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 ツリーを反映した、クリーンで Python 的な構造を保てます。

# 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 の完全実装

読み取り：タスクの表示（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'
    )

作成：新しいタスクの追加

作成フローには、HTMX 属性で拡張した HTML フォームを使用します。更新版では、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"

新しいタスクが追加されたことを確認できます。

画像：タスク追加後のスクリーンショット

更新：完了状態の切り替え

更新処理には 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))

削除：タスクの除去

削除は 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 フレームワークにありがちな複雑さを避けつつ、リアクティブなアプリケーションを構築できました。このアーキテクチャは、現代的な Web 開発においていくつかの重要な利点をもたらします。FastHTML のコンポーネントベース UI と MongoDB の柔軟なドキュメントモデルの相乗効果により、ビジネスロジック、データ整合性、プレゼンテーションを単一で一貫したエコシステム内で維持できます。この「オール Python」なアプローチは、開発のオーバーヘッドとデプロイの複雑さを大幅に削減します。