ライブラリ

MongoEngine

概要

MongoEngineはPythonでMongoDBを操作するためのObject-Document Mapper (ODM) です。リレーショナルデータベースのORMと同様の機能をドキュメント指向データベースに提供し、PythonオブジェクトとMongoDBドキュメント間のマッピングを直感的に実現します。Djangoライクなクエリ構文、強力なスキーマ定義、バリデーション機能、複雑なドキュメント構造のサポートを通じて、NoSQLアプリケーション開発を効率化。Django統合、マルチデータベース対応、インデックス管理など、企業レベルの要件に対応する包括的機能を提供します。

詳細

MongoEngine 2025年版はPythonでのMongoDB開発において最も成熟したODM選択肢として確立されています。Document、EmbeddedDocument、DynamicDocumentクラスによる柔軟なスキーマ設計と、StringField、IntField、ListField、ReferenceFieldなど豊富なフィールドタイプでドキュメント構造を明確に定義。Django風のQuerySetインターフェースにより、find、filter、aggregateといったMongoDB操作をPythonicに記述可能。リレーションシップ、継承、シグナル、カスタムマネージャーなど、ORM的機能をドキュメントデータベースに適用し、NoSQLの柔軟性とリレーショナルDBの構造化の利点を両立します。

主な特徴

• ドキュメント中心設計: Document、EmbeddedDocument、DynamicDocumentによる柔軟な構造定義
• 豊富なフィールドタイプ: String、Numeric、Date、Reference、List、MapFieldなど包括的対応
• Djangoライクなクエリ: filter、exclude、order_by等の直感的クエリ構文
• スキーマバリデーション: フィールドレベルでの型チェックと制約適用
• マルチデータベース対応: 複数MongoDB接続と動的データベース切り替え
• インデックス管理: パフォーマンス最適化のための柔軟なインデックス定義

メリット・デメリット

メリット

• PythonとMongoDBの理想的な組み合わせで学習コストが低い
• Djangoエコシステムとの優れた統合とDjango REST frameworkサポート
• スキーマレスとスキーマありの柔軟な選択（DynamicDocument活用）
• リッチなドキュメント操作とネストされた構造の直感的扱い
• 包括的なクエリ機能とaggregationパイプライン対応
• マルチテナンシーとマルチデータベース環境での優れた拡張性

デメリット

• MongoDBでの高度なトランザクション処理は制限的
• リレーショナルJOINのような複雑な関連操作は不得意
• 大量データでのパフォーマンスチューニングには深いMongoDB知識が必要
• マイグレーション機能がリレーショナルORMより限定的
• スキーマ変更時の既存データ互換性管理が困難
• MongoDB固有の機能に依存しているため他のデータベースへの移行が不可

参考ページ

• MongoEngine 公式ドキュメント
• MongoEngine GitHub リポジトリ
• MongoEngine API リファレンス

書き方の例

インストールと基本セットアップ

# MongoEngineのインストール
pip install mongoengine

# 依存関係とMongoDB用ライブラリ
pip install pymongo dnspython

# MongoDBサーバーの起動（Docker使用例）
docker run -d -p 27017:27017 --name mongodb mongo:latest

# Python環境での動作確認
python -c "import mongoengine; print(mongoengine.__version__)"

# settings.py または main.py での基本接続設定
import mongoengine

# 基本接続（ローカルMongoDB）
mongoengine.connect('my_database')

# 詳細接続設定
mongoengine.connect(
    db='my_database',
    host='localhost',
    port=27017,
    username='myuser',
    password='mypassword',
    authentication_source='admin'
)

# MongoDB URI形式での接続
mongoengine.connect(
    host='mongodb://myuser:mypassword@localhost:27017/my_database?authSource=admin'
)

# 複数データベース接続
mongoengine.connect('main_db', alias='default')
mongoengine.connect('logs_db', alias='logs')

ドキュメント定義と基本的なスキーマ

from mongoengine import Document, EmbeddedDocument, DynamicDocument
from mongoengine import StringField, IntField, DateTimeField, EmailField
from mongoengine import ListField, EmbeddedDocumentField, ReferenceField
from datetime import datetime

# 基本的なドキュメント定義
class User(Document):
    username = StringField(required=True, unique=True, max_length=50)
    email = EmailField(required=True, unique=True)
    password = StringField(required=True, min_length=8)
    first_name = StringField(max_length=50)
    last_name = StringField(max_length=50)
    created_at = DateTimeField(default=datetime.utcnow)
    is_active = BooleanField(default=True)
    
    # メタ設定
    meta = {
        'collection': 'users',
        'indexes': [
            'username',
            'email',
            ('created_at', -1),  # 降順インデックス
        ],
        'ordering': ['-created_at']
    }
    
    def __str__(self):
        return f"{self.username} ({self.email})"

# 埋め込みドキュメント
class Address(EmbeddedDocument):
    street = StringField(max_length=100)
    city = StringField(max_length=50)
    country = StringField(max_length=50)
    postal_code = StringField(max_length=20)

# リッチなドキュメント構造
class BlogPost(Document):
    title = StringField(required=True, max_length=200)
    slug = StringField(required=True, unique=True)
    content = StringField()
    author = ReferenceField(User, required=True)
    tags = ListField(StringField(max_length=30))
    published = BooleanField(default=False)
    created_at = DateTimeField(default=datetime.utcnow)
    updated_at = DateTimeField(default=datetime.utcnow)
    
    # 埋め込みドキュメント
    author_info = EmbeddedDocumentField(Address)
    
    meta = {
        'collection': 'blog_posts',
        'indexes': [
            'slug',
            'author',
            ('published', 'created_at'),
            {
                'fields': ['$title', '$content'],  # テキストインデックス
                'default_language': 'english'
            }
        ]
    }

# 動的スキーマ（スキーマレス）
class DynamicPost(DynamicDocument):
    title = StringField(required=True)
    # その他のフィールドは動的に追加可能
    
    meta = {
        'collection': 'dynamic_posts'
    }

基本的なCRUD操作

from datetime import datetime

# Create - ドキュメント作成
user = User(
    username='john_doe',
    email='[email protected]',
    password='secure_password',
    first_name='John',
    last_name='Doe'
)
user.save()

# または辞書形式での作成
user = User.objects.create(
    username='jane_doe',
    email='[email protected]',
    password='another_password'
)

# Read - ドキュメント取得
# 全件取得
all_users = User.objects.all()

# 条件指定取得
user = User.objects.get(username='john_doe')
active_users = User.objects.filter(is_active=True)

# 複数条件
recent_users = User.objects.filter(
    is_active=True,
    created_at__gte=datetime(2024, 1, 1)
)

# クエリ演算子
users = User.objects.filter(
    username__icontains='john',  # 部分一致（大文字小文字無視）
    created_at__lt=datetime.now(),  # より小さい
    email__endswith='@example.com'  # 末尾一致
)

# Update - ドキュメント更新
user = User.objects.get(username='john_doe')
user.first_name = 'Jonathan'
user.save()

# 一括更新
User.objects.filter(is_active=False).update(set__is_active=True)

# アトミック更新
User.objects.filter(username='john_doe').update(
    inc__login_count=1,  # インクリメント
    set__last_login=datetime.utcnow()
)

# Delete - ドキュメント削除
user = User.objects.get(username='john_doe')
user.delete()

# 一括削除
User.objects.filter(is_active=False).delete()

高度なクエリとアグリゲーション

from mongoengine import Q
from mongoengine.queryset import QuerySet

# Q オブジェクトを使った複雑なクエリ
complex_query = User.objects.filter(
    Q(username__icontains='admin') | Q(email__endswith='@admin.com')
)

# ソートと制限
users = User.objects.filter(is_active=True)\
    .order_by('-created_at')\
    .limit(10)\
    .skip(20)  # ページネーション

# 射影（特定フィールドのみ取得）
user_emails = User.objects.only('username', 'email')
user_summary = User.objects.exclude('password')

# ドキュメント数カウント
user_count = User.objects.filter(is_active=True).count()

# 存在チェック
exists = User.objects.filter(username='admin').first() is not None

# アグリゲーションパイプライン
from mongoengine import Q

# グループ化と集計
pipeline = [
    {'$match': {'published': True}},
    {'$group': {
        '_id': '$author',
        'post_count': {'$sum': 1},
        'latest_post': {'$max': '$created_at'}
    }},
    {'$sort': {'post_count': -1}}
]

results = BlogPost.objects.aggregate(pipeline)

# 生のMongoDBクエリ実行
raw_query = {'username': {'$regex': '^admin', '$options': 'i'}}
users = User.objects(__raw__=raw_query)

リレーションシップと参照

# ReferenceField を使った関連
class Comment(Document):
    content = StringField(required=True)
    author = ReferenceField(User, required=True)
    post = ReferenceField(BlogPost, required=True)
    created_at = DateTimeField(default=datetime.utcnow)
    
    meta = {
        'collection': 'comments',
        'indexes': [
            'post',
            'author',
            ('post', 'created_at')
        ]
    }

# 関連ドキュメントの作成
user = User.objects.get(username='john_doe')
post = BlogPost.objects.get(slug='my-first-post')

comment = Comment(
    content='Great post!',
    author=user,
    post=post
)
comment.save()

# 逆参照クエリ
# 特定ユーザーのすべてのコメント
user_comments = Comment.objects.filter(author=user)

# 特定投稿のすべてのコメント
post_comments = Comment.objects.filter(post=post)

# populate（JOINライクな操作）
comments_with_authors = Comment.objects.select_related('author', 'post')

for comment in comments_with_authors:
    print(f"{comment.author.username}: {comment.content}")
    print(f"Post: {comment.post.title}")

# ListFieldの参照
class Tag(Document):
    name = StringField(required=True, unique=True)
    color = StringField(default='#000000')

class Post(Document):
    title = StringField(required=True)
    tags = ListField(ReferenceField(Tag))
    
# タグ付き投稿の作成
tech_tag = Tag.objects.create(name='Technology', color='#00ff00')
python_tag = Tag.objects.create(name='Python', color='#0000ff')

post = Post.objects.create(
    title='Python Tutorial',
    tags=[tech_tag, python_tag]
)

# タグでの検索
tech_posts = Post.objects.filter(tags=tech_tag)

継承とポリモーフィズム

# ドキュメント継承
class Animal(Document):
    name = StringField(required=True)
    species = StringField(required=True)
    
    meta = {
        'collection': 'animals',
        'allow_inheritance': True  # 継承を許可
    }

class Dog(Animal):
    breed = StringField()
    is_good_boy = BooleanField(default=True)

class Cat(Animal):
    indoor = BooleanField(default=True)
    lives_remaining = IntField(default=9)

# 継承されたドキュメントの操作
dog = Dog.objects.create(
    name='Buddy',
    species='Canis lupus',
    breed='Golden Retriever'
)

cat = Cat.objects.create(
    name='Whiskers',
    species='Felis catus',
    indoor=True
)

# ポリモーフィッククエリ
all_animals = Animal.objects.all()  # Dog と Cat の両方を含む

# 特定のサブクラスのみ
dogs_only = Dog.objects.all()
cats_only = Cat.objects.all()

# 型チェック
for animal in all_animals:
    if isinstance(animal, Dog):
        print(f"Dog: {animal.name}, Breed: {animal.breed}")
    elif isinstance(animal, Cat):
        print(f"Cat: {animal.name}, Lives: {animal.lives_remaining}")

カスタムフィールドとバリデーション

from mongoengine import ValidationError
import re

# カスタムバリデーション関数
def validate_phone_number(phone):
    pattern = r'^\+?1?\d{9,15}$'
    if not re.match(pattern, phone):
        raise ValidationError('Invalid phone number format')

# カスタムフィールド
class PhoneNumberField(StringField):
    def validate(self, value):
        super().validate(value)
        validate_phone_number(value)

# バリデーション付きドキュメント
class Contact(Document):
    name = StringField(required=True, min_length=2, max_length=100)
    email = EmailField(required=True)
    phone = PhoneNumberField()
    age = IntField(min_value=0, max_value=150)
    
    def validate(self):
        # ドキュメントレベルのバリデーション
        if self.age and self.age < 18 and not self.parent_email:
            raise ValidationError('Minors must provide parent email')
    
    def clean(self):
        # 保存前の自動クリーニング
        if self.email:
            self.email = self.email.lower()

# カスタムマネージャー
class ActiveUserManager:
    def get_queryset(self):
        return User.objects.filter(is_active=True)

class User(Document):
    # ... フィールド定義 ...
    
    # カスタムマネージャー
    active_objects = ActiveUserManager()
    
    @classmethod
    def create_user(cls, username, email, password):
        """カスタム作成メソッド"""
        user = cls(
            username=username,
            email=email.lower(),
            password=hash_password(password)
        )
        user.save()
        return user

マルチデータベースとコンテキスト管理

import mongoengine
from mongoengine.context_managers import switch_db

# 複数データベース接続
mongoengine.connect('main_db', alias='default')
mongoengine.connect('analytics_db', alias='analytics')
mongoengine.connect('logs_db', alias='logs')

# データベース別のドキュメント定義
class User(Document):
    username = StringField(required=True)
    email = EmailField(required=True)
    
    meta = {
        'db_alias': 'default'  # メインデータベース
    }

class UserAnalytics(Document):
    user_id = StringField(required=True)
    page_views = IntField(default=0)
    last_activity = DateTimeField()
    
    meta = {
        'db_alias': 'analytics'  # 分析データベース
    }

class SystemLog(Document):
    level = StringField(choices=['INFO', 'WARNING', 'ERROR'])
    message = StringField(required=True)
    timestamp = DateTimeField(default=datetime.utcnow)
    
    meta = {
        'db_alias': 'logs'  # ログデータベース
    }

# データベース切り替え
with switch_db(User, 'analytics'):
    # 一時的にanalyticsデータベースに切り替え
    analytics_user = User.objects.create(username='temp_user')

# 通常のデータベース操作
main_user = User.objects.create(username='main_user')

# 異なるデータベースでの並行操作
user = User.objects.create(username='john_doe')
analytics = UserAnalytics.objects.create(
    user_id=str(user.id),
    page_views=1
)
log = SystemLog.objects.create(
    level='INFO',
    message=f'User {user.username} created'
)