Library

MongoEngine

Overview

MongoEngine is an Object-Document Mapper (ODM) for operating MongoDB with Python. It provides functionality similar to relational database ORMs for document-oriented databases, intuitively realizing mapping between Python objects and MongoDB documents. Through Django-like query syntax, powerful schema definition, validation features, and support for complex document structures, it streamlines NoSQL application development. It provides comprehensive functionality for enterprise-level requirements including Django integration, multi-database support, and index management.

Details

MongoEngine 2025 version is established as the most mature ODM choice for MongoDB development in Python. Flexible schema design through Document, EmbeddedDocument, and DynamicDocument classes, and clear document structure definition with rich field types including StringField, IntField, ListField, and ReferenceField. MongoDB operations can be written Pythonically through Django-style QuerySet interface with find, filter, and aggregate operations. By applying ORM-like features such as relationships, inheritance, signals, and custom managers to document databases, it achieves the benefits of both NoSQL flexibility and relational DB structure.

Key Features

• Document-Centered Design: Flexible structure definition through Document, EmbeddedDocument, and DynamicDocument
• Rich Field Types: Comprehensive support including String, Numeric, Date, Reference, List, MapField
• Django-like Queries: Intuitive query syntax with filter, exclude, order_by
• Schema Validation: Type checking and constraint application at field level
• Multi-Database Support: Multiple MongoDB connections and dynamic database switching
• Index Management: Flexible index definition for performance optimization

Pros and Cons

Pros

• Low learning cost with ideal combination of Python and MongoDB
• Excellent integration with Django ecosystem and Django REST framework support
• Flexible choice between schema-less and schema-based (utilizing DynamicDocument)
• Intuitive handling of rich document operations and nested structures
• Comprehensive query capabilities and aggregation pipeline support
• Excellent scalability in multi-tenancy and multi-database environments

Cons

• Advanced transaction processing in MongoDB is restrictive
• Complex relational operations like relational JOINs are not well-suited
• Deep MongoDB knowledge required for performance tuning with large data
• Migration features are more limited than relational ORMs
• Difficult to manage existing data compatibility when schema changes
• Cannot migrate to other databases due to dependency on MongoDB-specific features

Reference Pages

• MongoEngine Official Documentation
• MongoEngine GitHub Repository
• MongoEngine API Reference

Code Examples

Installation and Basic Setup

# Install MongoEngine
pip install mongoengine

# Dependencies and MongoDB libraries
pip install pymongo dnspython

# Start MongoDB server (Docker example)
docker run -d -p 27017:27017 --name mongodb mongo:latest

# Verify in Python environment
python -c "import mongoengine; print(mongoengine.__version__)"

# Basic connection settings in settings.py or main.py
import mongoengine

# Basic connection (local MongoDB)
mongoengine.connect('my_database')

# Detailed connection settings
mongoengine.connect(
    db='my_database',
    host='localhost',
    port=27017,
    username='myuser',
    password='mypassword',
    authentication_source='admin'
)

# MongoDB URI format connection
mongoengine.connect(
    host='mongodb://myuser:mypassword@localhost:27017/my_database?authSource=admin'
)

# Multiple database connections
mongoengine.connect('main_db', alias='default')
mongoengine.connect('logs_db', alias='logs')

Document Definition and Basic Schema

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

# Basic document definition
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 settings
    meta = {
        'collection': 'users',
        'indexes': [
            'username',
            'email',
            ('created_at', -1),  # Descending index
        ],
        'ordering': ['-created_at']
    }
    
    def __str__(self):
        return f"{self.username} ({self.email})"

# Embedded document
class Address(EmbeddedDocument):
    street = StringField(max_length=100)
    city = StringField(max_length=50)
    country = StringField(max_length=50)
    postal_code = StringField(max_length=20)

# Rich document structure
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)
    
    # Embedded document
    author_info = EmbeddedDocumentField(Address)
    
    meta = {
        'collection': 'blog_posts',
        'indexes': [
            'slug',
            'author',
            ('published', 'created_at'),
            {
                'fields': ['$title', '$content'],  # Text index
                'default_language': 'english'
            }
        ]
    }

# Dynamic schema (schema-less)
class DynamicPost(DynamicDocument):
    title = StringField(required=True)
    # Other fields can be added dynamically
    
    meta = {
        'collection': 'dynamic_posts'
    }

Basic CRUD Operations

from datetime import datetime

# Create - Document creation
user = User(
    username='john_doe',
    email='[email protected]',
    password='secure_password',
    first_name='John',
    last_name='Doe'
)
user.save()

# Or create using dictionary format
user = User.objects.create(
    username='jane_doe',
    email='[email protected]',
    password='another_password'
)

# Read - Document retrieval
# Get all documents
all_users = User.objects.all()

# Conditional retrieval
user = User.objects.get(username='john_doe')
active_users = User.objects.filter(is_active=True)

# Multiple conditions
recent_users = User.objects.filter(
    is_active=True,
    created_at__gte=datetime(2024, 1, 1)
)

# Query operators
users = User.objects.filter(
    username__icontains='john',  # Partial match (case insensitive)
    created_at__lt=datetime.now(),  # Less than
    email__endswith='@example.com'  # Ends with
)

# Update - Document update
user = User.objects.get(username='john_doe')
user.first_name = 'Jonathan'
user.save()

# Bulk update
User.objects.filter(is_active=False).update(set__is_active=True)

# Atomic update
User.objects.filter(username='john_doe').update(
    inc__login_count=1,  # Increment
    set__last_login=datetime.utcnow()
)

# Delete - Document deletion
user = User.objects.get(username='john_doe')
user.delete()

# Bulk deletion
User.objects.filter(is_active=False).delete()

Advanced Queries and Aggregation

from mongoengine import Q
from mongoengine.queryset import QuerySet

# Complex queries using Q objects
complex_query = User.objects.filter(
    Q(username__icontains='admin') | Q(email__endswith='@admin.com')
)

# Sorting and limiting
users = User.objects.filter(is_active=True)\
    .order_by('-created_at')\
    .limit(10)\
    .skip(20)  # Pagination

# Projection (retrieve specific fields only)
user_emails = User.objects.only('username', 'email')
user_summary = User.objects.exclude('password')

# Document count
user_count = User.objects.filter(is_active=True).count()

# Existence check
exists = User.objects.filter(username='admin').first() is not None

# Aggregation pipeline
from mongoengine import Q

# Grouping and aggregation
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)

# Raw MongoDB query execution
raw_query = {'username': {'$regex': '^admin', '$options': 'i'}}
users = User.objects(__raw__=raw_query)

Relationships and References

# Using ReferenceField for relations
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')
        ]
    }

# Creating related documents
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()

# Reverse reference queries
# All comments by specific user
user_comments = Comment.objects.filter(author=user)

# All comments on specific post
post_comments = Comment.objects.filter(post=post)

# Populate (JOIN-like operation)
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 references
class Tag(Document):
    name = StringField(required=True, unique=True)
    color = StringField(default='#000000')

class Post(Document):
    title = StringField(required=True)
    tags = ListField(ReferenceField(Tag))
    
# Creating tagged posts
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]
)

# Search by tags
tech_posts = Post.objects.filter(tags=tech_tag)

Inheritance and Polymorphism

# Document inheritance
class Animal(Document):
    name = StringField(required=True)
    species = StringField(required=True)
    
    meta = {
        'collection': 'animals',
        'allow_inheritance': True  # Allow inheritance
    }

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

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

# Operating inherited documents
dog = Dog.objects.create(
    name='Buddy',
    species='Canis lupus',
    breed='Golden Retriever'
)

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

# Polymorphic queries
all_animals = Animal.objects.all()  # Includes both Dogs and Cats

# Specific subclass only
dogs_only = Dog.objects.all()
cats_only = Cat.objects.all()

# Type checking
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}")

Custom Fields and Validation

from mongoengine import ValidationError
import re

# Custom validation function
def validate_phone_number(phone):
    pattern = r'^\+?1?\d{9,15}$'
    if not re.match(pattern, phone):
        raise ValidationError('Invalid phone number format')

# Custom field
class PhoneNumberField(StringField):
    def validate(self, value):
        super().validate(value)
        validate_phone_number(value)

# Document with validation
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):
        # Document level validation
        if self.age and self.age < 18 and not self.parent_email:
            raise ValidationError('Minors must provide parent email')
    
    def clean(self):
        # Auto-cleaning before save
        if self.email:
            self.email = self.email.lower()

# Custom manager
class ActiveUserManager:
    def get_queryset(self):
        return User.objects.filter(is_active=True)

class User(Document):
    # ... field definitions ...
    
    # Custom manager
    active_objects = ActiveUserManager()
    
    @classmethod
    def create_user(cls, username, email, password):
        """Custom creation method"""
        user = cls(
            username=username,
            email=email.lower(),
            password=hash_password(password)
        )
        user.save()
        return user

Multi-Database and Context Management

import mongoengine
from mongoengine.context_managers import switch_db

# Multiple database connections
mongoengine.connect('main_db', alias='default')
mongoengine.connect('analytics_db', alias='analytics')
mongoengine.connect('logs_db', alias='logs')

# Database-specific document definitions
class User(Document):
    username = StringField(required=True)
    email = EmailField(required=True)
    
    meta = {
        'db_alias': 'default'  # Main database
    }

class UserAnalytics(Document):
    user_id = StringField(required=True)
    page_views = IntField(default=0)
    last_activity = DateTimeField()
    
    meta = {
        'db_alias': 'analytics'  # Analytics database
    }

class SystemLog(Document):
    level = StringField(choices=['INFO', 'WARNING', 'ERROR'])
    message = StringField(required=True)
    timestamp = DateTimeField(default=datetime.utcnow)
    
    meta = {
        'db_alias': 'logs'  # Log database
    }

# Database switching
with switch_db(User, 'analytics'):
    # Temporarily switch to analytics database
    analytics_user = User.objects.create(username='temp_user')

# Normal database operations
main_user = User.objects.create(username='main_user')

# Concurrent operations on different databases
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'
)