Mongoengine - work with Mongodb

Complete Guide to MongoEngine: Working with MongoDB in Python

Introduction to MongoEngine

As NoSQL databases evolve, more Python developers choose MongoDB for data storage. This is due to its flexibility, scalability, and document‑oriented structure. To simplify interaction with MongoDB, the MongoEngine library was created – an object‑document mapper (ODM) that lets you work with MongoDB documents as Python objects.

MongoEngine offers a powerful yet straightforward interface for defining models, executing queries, and validating data without the need to write complex MongoDB‑specific code. It is one of the most popular solutions for working with MongoDB in the Python ecosystem.

MongoDB Basics and the Role of MongoEngine

NoSQL and Document‑Oriented Storage

MongoDB is a document‑oriented database where data is stored in BSON (binary JSON) format. This makes MongoDB flexible, especially when dealing with complex nested data structures. Unlike relational databases, MongoDB does not require a rigid schema and allows dynamic changes to document structures.

Object‑Document Mapping (ODM)

An ODM works similarly to an ORM, but instead of tables and rows it operates on documents and collections. MongoEngine implements ODM abstractions for MongoDB, providing a clean and Pythonic API. This enables developers to interact with the database using familiar Python objects without diving into MongoDB syntax details.

Benefits of MongoEngine

MongoEngine offers several key advantages:

• Declarative syntax: Models are defined as Python classes
• Automatic validation: Built‑in field‑level data checks
• Relationship support: Ability to create links between collections
• Indexing: Simple definition of indexes for query optimization
• Web‑framework integration: Easy integration with Flask, Django, and others

Installing and Configuring MongoEngine

Library Installation

pip install mongoengine

To work with MongoDB you also need to install the MongoDB server itself. You can use a local installation or cloud services such as MongoDB Atlas.

Connecting to the Database

Basic connection to a local MongoDB instance:

from mongoengine import connect

connect(db='library', host='localhost', port=27017)

Connection using a connection string:

connect(host='mongodb://localhost:27017/library')

Connecting to MongoDB Atlas or another cloud service:

connect(host='mongodb+srv://username:password@cluster.mongodb.net/database?retryWrites=true&w=majority')

Authenticated Connection Settings

connect(
    db='library',
    host='localhost',
    port=27017,
    username='admin',
    password='password',
    authentication_source='admin'
)

Creating Models with MongoEngine

Model Definition Basics

MongoEngine uses Python classes to describe document structures. Each model inherits from the Document class:

from mongoengine import Document, StringField, DateField, IntField

class Author(Document):
    name = StringField(required=True, max_length=100)
    birth_date = DateField()
    books_count = IntField(default=0)
    
    def __str__(self):
        return self.name

Model Metadata

The Meta class lets you configure various model parameters:

class Book(Document):
    title = StringField(required=True)
    isbn = StringField(unique=True)
    pages = IntField()
    
    meta = {
        'collection': 'books',
        'indexes': ['isbn', 'title'],
        'ordering': ['-title']
    }

Field Types and Options

MongoEngine provides a rich set of field types:

from mongoengine import *

class ComplexModel(Document):
    # String fields
    name = StringField(required=True, max_length=50)
    email = EmailField()
    url = URLField()
    
    # Numeric fields
    age = IntField(min_value=0, max_value=150)
    price = FloatField()
    score = DecimalField(precision=2)
    
    # Boolean and temporal fields
    is_active = BooleanField(default=True)
    created_at = DateTimeField(default=datetime.now)
    birth_date = DateField()
    
    # File fields
    avatar = ImageField()
    document = FileField()
    
    # Special fields
    tags = ListField(StringField(max_length=30))
    metadata = DictField()
    coordinates = PointField()

CRUD Operations in MongoEngine

Creating Documents

# Create and save a document
author = Author(name="Leo Tolstoy", birth_date=datetime(1828, 9, 9))
author.save()

# Alternative approach
author = Author.objects.create(
    name="Fyodor Dostoevsky",
    birth_date=datetime(1821, 11, 11)
)

Reading Documents

# Retrieve all documents
authors = Author.objects()

# Retrieve a single document
author = Author.objects.get(name="Leo Tolstoy")

# Safe retrieval
author = Author.objects.filter(name="Leo Tolstoy").first()

# Retrieval with conditions
young_authors = Author.objects(birth_date__gte=datetime(1900, 1, 1))

Updating Documents

# Update a specific document
author = Author.objects.get(name="Leo Tolstoy")
author.books_count = 15
author.save()

# Atomic update
author.update(inc__books_count=1)

# Bulk update
Author.objects(birth_date__lt=datetime(1900, 1, 1)).update(
    set__is_classic=True
)

Deleting Documents

# Delete a specific document
author = Author.objects.get(name="Leo Tolstoy")
author.delete()

# Bulk delete
Author.objects(books_count=0).delete()

Advanced Queries and Filtering

Query Operators

MongoEngine supports many operators for complex queries:

# Comparison
Author.objects(age__gt=30)      # greater than 30
Author.objects(age__gte=30)     # greater than or equal to 30
Author.objects(age__lt=50)      # less than 50
Author.objects(age__lte=50)     # less than or equal to 50

# Inclusion
Author.objects(name__in=['Tolstoy', 'Dostoevsky'])
Author.objects(name__not__in=['Unknown'])

# Substring search
Author.objects(name__contains='Leo')
Author.objects(name__icontains='leo')  # case‑insensitive

# Regular expressions
Author.objects(name__regex=r'^A.*')

# Field existence check
Author.objects(email__exists=True)

Sorting and Limiting Results

# Sorting
authors = Author.objects.order_by('name')          # ascending
authors = Author.objects.order_by('-birth_date')  # descending

# Limiting result count
authors = Author.objects[:10]      # first 10
authors = Author.objects[10:20]    # 10 through 20

# Pagination
page = 1
per_page = 10
authors = Author.objects.skip((page - 1) * per_page).limit(per_page)

Projections and Field Exclusions

# Include only selected fields
authors = Author.objects.only('name', 'birth_date')

# Exclude specific fields
authors = Author.objects.exclude('metadata', 'description')

# Exclude fields from embedded documents
authors = Author.objects.exclude('books__content')

Working with Embedded Documents

Defining Embedded Documents

from mongoengine import EmbeddedDocument, EmbeddedDocumentField, ListField

class Address(EmbeddedDocument):
    street = StringField(required=True)
    city = StringField(required=True)
    postal_code = StringField()
    country = StringField(default='Russia')
    
    def __str__(self):
        return f"{self.street}, {self.city}"

class ContactInfo(EmbeddedDocument):
    email = EmailField()
    phone = StringField()
    website = URLField()

class Publisher(Document):
    name = StringField(required=True)
    address = EmbeddedDocumentField(Address)
    contact = EmbeddedDocumentField(ContactInfo)
    offices = ListField(EmbeddedDocumentField(Address))

Manipulating Embedded Documents

# Create with embedded documents
publisher = Publisher(
    name="AST Publishing",
    address=Address(
        street="Star Boulevard, 21",
        city="Moscow",
        postal_code="129085"
    ),
    contact=ContactInfo(
        email="info@ast.ru",
        phone="+7 495 615-53-10"
    )
)
publisher.save()

# Queries on embedded documents
moscow_publishers = Publisher.objects(address__city="Moscow")
publishers_with_email = Publisher.objects(contact__email__exists=True)

Data Validation and Constraints

Built‑in Validation

MongoEngine provides many built‑in validators:

class User(Document):
    username = StringField(
        required=True,
        min_length=3,
        max_length=20,
        unique=True,
        regex=r'^[a-zA-Z0-9_]+$'
    )
    email = EmailField(required=True, unique=True)
    age = IntField(min_value=13, max_value=120)
    password = StringField(min_length=8)
    
    # Custom validation
    def clean(self):
        if self.username.lower() in ['admin', 'root', 'administrator']:
            raise ValidationError('Invalid username')
        
        if self.age < 18 and not self.parent_consent:
            raise ValidationError('Parental consent required')

Custom Validators

def validate_isbn(isbn):
    if len(isbn) not in [10, 13]:
        raise ValidationError('ISBN must contain 10 or 13 digits')
    
    if not isbn.isdigit():
        raise ValidationError('ISBN must contain only numbers')

class Book(Document):
    title = StringField(required=True)
    isbn = StringField(validation=validate_isbn)
    
    def clean(self):
        # Document‑level validation
        if self.publication_date > datetime.now():
            raise ValidationError('Publication date cannot be in the future')

Relationships Between Documents

Types of Relationships

MongoEngine supports several relationship types:

from mongoengine import ReferenceField, LazyReferenceField, GenericReferenceField

class Author(Document):
    name = StringField(required=True)
    
class Genre(Document):
    name = StringField(required=True)
    
class Book(Document):
    title = StringField(required=True)
    # Standard reference
    author = ReferenceField(Author, required=True)
    # Lazy reference (loaded on access)
    genre = LazyReferenceField(Genre)
    # List of references
    co_authors = ListField(ReferenceField(Author))
    # Reverse reference
    reviews = ListField(ReferenceField('Review'))

class Review(Document):
    book = ReferenceField(Book, required=True)
    rating = IntField(min_value=1, max_value=5)
    comment = StringField()

Deletion Rules for Related Objects

class Author(Document):
    name = StringField(required=True)

class Book(Document):
    title = StringField(required=True)
    author = ReferenceField(
        Author,
        reverse_delete_rule=CASCADE  # Delete books when author is removed
    )

class Review(Document):
    book = ReferenceField(
        Book,
        reverse_delete_rule=CASCADE  # Delete reviews when book is removed
    )
    user = ReferenceField(
        User,
        reverse_delete_rule=NULLIFY  # Nullify field when user is removed
    )

Working with Related Documents

# Create related documents
author = Author(name="Alexander Pushkin")
author.save()

book = Book(title="Eugene Onegin", author=author)
book.save()

# Retrieve related documents
books_by_pushkin = Book.objects(author=author)

# Load related documents efficiently
book = Book.objects.select_related('author').first()
print(book.author.name)  # No extra DB query

Indexing and Performance

Creating Indexes

class Book(Document):
    title = StringField(required=True)
    isbn = StringField(unique=True)  # Automatically creates a unique index
    author = ReferenceField(Author)
    publication_date = DateTimeField()
    
    meta = {
        'indexes': [
            'isbn',                                 # Simple index
            ('author', 'publication_date'),         # Compound index
            {
                'fields': ['title'],
                'unique': True,
                'sparse': True
            },                                      # Index with options
            {
                'fields': ['$title', '$author.name'],  # Text index
                'default_language': 'russian'
            }
        ]
    }

Query Optimization

# Use explain() to analyze queries
query = Book.objects(author=author)
print(query.explain())

# Prefetch related documents
books = Book.objects.select_related('author', 'genre')

# Use projections to reduce traffic
books = Book.objects.only('title', 'author')

# Use limit() and skip() for pagination
books = Book.objects.skip(20).limit(10)

Integration with Web Frameworks

Flask Integration

from flask import Flask, jsonify, request
from mongoengine import connect
from flask_mongoengine import MongoEngine

app = Flask(__name__)
app.config['MONGODB_SETTINGS'] = {
    'db': 'library',
    'host': 'localhost',
    'port': 27017
}

db = MongoEngine(app)

class Book(db.Document):
    title = db.StringField(required=True)
    author = db.StringField(required=True)
    
    def to_dict(self):
        return {
            'id': str(self.id),
            'title': self.title,
            'author': self.author
        }

@app.route('/books', methods=['GET'])
def get_books():
    books = Book.objects()
    return jsonify([book.to_dict() for book in books])

@app.route('/books', methods=['POST'])
def create_book():
    data = request.get_json()
    book = Book(title=data['title'], author=data['author'])
    book.save()
    return jsonify(book.to_dict()), 201

FastAPI Integration

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from mongoengine import connect
from typing import List, Optional

app = FastAPI()
connect('library')

class BookCreate(BaseModel):
    title: str
    author: str

class BookResponse(BaseModel):
    id: str
    title: str
    author: str

@app.post("/books/", response_model=BookResponse)
async def create_book(book: BookCreate):
    db_book = Book(title=book.title, author=book.author)
    db_book.save()
    return BookResponse(
        id=str(db_book.id),
        title=db_book.title,
        author=db_book.author
    )

@app.get("/books/", response_model=List[BookResponse])
async def get_books():
    books = Book.objects()
    return [
        BookResponse(
            id=str(book.id),
            title=book.title,
            author=book.author
        ) for book in books
    ]

Working with Transactions and Sessions

MongoDB Transactions

MongoEngine supports transactions starting with MongoDB 4.0:

from mongoengine import get_db
from pymongo import WriteConcern

def transfer_book():
    with get_db().client.start_session() as session:
        with session.start_transaction():
            # Decrease book count in the source library
            Library.objects(id=source_library_id).update_one(
                dec__books_count=1,
                session=session
            )
            
            # Increase book count in the target library
            Library.objects(id=target_library_id).update_one(
                inc__books_count=1,
                session=session
            )
            
            # Update the book record
            Book.objects(id=book_id).update_one(
                set__library=target_library_id,
                session=session
            )

Using Sessions

from mongoengine import get_db

def batch_operation():
    db = get_db()
    with db.client.start_session() as session:
        with session.start_transaction():
            # All operations run within a single transaction
            for book_data in books_data:
                book = Book(**book_data)
                book.save(session=session)

Testing with MongoEngine

Using mongomock

pip install mongomock

import mongomock
from mongoengine import connect, disconnect

def setup_test_db():
    disconnect()
    connect('testdb', host='mongomock://localhost')

def teardown_test_db():
    disconnect()

# Example test
def test_book_creation():
    setup_test_db()
    
    book = Book(title="Test Book", author="Test Author")
    book.save()
    
    assert Book.objects.count() == 1
    assert Book.objects.first().title == "Test Book"
    
    teardown_test_db()

Testing with pytest

import pytest
from mongoengine import connect, disconnect

@pytest.fixture
def db_connection():
    connect('testdb', host='mongomock://localhost')
    yield
    disconnect()

def test_book_validation(db_connection):
    with pytest.raises(ValidationError):
        book = Book(title="")  # Empty title should raise an error
        book.save()

Comprehensive Table of MongoEngine Methods and Functions

Query Operators in MongoEngine

Advanced Features of MongoEngine

Signals

MongoEngine supports a signal system to perform actions on specific events:

from mongoengine import signals

class User(Document):
    name = StringField()
    email = EmailField()
    created_at = DateTimeField()
    updated_at = DateTimeField()

@signals.pre_save.connect
def update_timestamps(sender, document, **kwargs):
    if not document.created_at:
        document.created_at = datetime.now()
    document.updated_at = datetime.now()

@signals.post_save.connect
def send_welcome_email(sender, document, created, **kwargs):
    if created and isinstance(document, User):
        # Send welcome email
        send_email(document.email, "Welcome!")

Custom Fields

class UpperCaseStringField(StringField):
    def validate(self, value):
        super().validate(value)
        if value and not value.isupper():
            raise ValidationError('Value must be uppercase')
    
    def to_mongo(self, value):
        return value.upper() if value else value

class User(Document):
    name = StringField()
    department_code = UpperCaseStringField()

Working with GridFS

from mongoengine import GridFSProxy
from gridfs import GridFS

class Document(Document):
    title = StringField()
    file = FileField()
    
    def save_file(self, file_path):
        with open(file_path, 'rb') as f:
            self.file.put(f, content_type='application/pdf')
        self.save()
    
    def get_file_content(self):
        return self.file.read()

Comparison with Other ODMs and Drivers

Performance Optimization

Best Practices

1. Use indexes for frequently queried fields
2. Limit result sets with limit()
3. Apply projections to load only required fields
4. Prefetch relationships using select_related()
5. Leverage aggregation pipelines for complex calculations

Query Monitoring

import logging

# Enable query logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('mongoengine')

# Analyze query performance
def analyze_query_performance():
    import time
    
    start_time = time.time()
    users = User.objects(age__gt=18)
    query_time = time.time() - start_time
    
    print(f"Query execution time: {query_time:.4f} seconds")
    print(f"Result count: {users.count()}")
    print(f"Execution plan: {users.explain()}")

The Future of MongoEngine and Project Activity

MongoEngine is a mature and stable library that receives active community support. Key development directions include:

• Performance improvements
• Better support for newer MongoDB versions
• Enhanced aggregation capabilities
• Improved integration with asynchronous frameworks
• More flexible validation system

The project is hosted on GitHub at: https://github.com/MongoEngine/mongoengine

Frequently Asked Questions

What is MongoEngine and what is it used for?

MongoEngine is an object‑document mapper (ODM) for working with MongoDB in Python. It lets you treat MongoDB documents as regular Python objects, providing a convenient API for model creation, querying, and data validation.

Does MongoEngine support relationships between collections?

Yes, MongoEngine supports various relationship types: ReferenceField for simple references, LazyReferenceField for lazy loading, GenericReferenceField for generic references, and ListField(ReferenceField) for one‑to‑many relationships.

Can I use MongoEngine with Flask?

Absolutely. MongoEngine integrates easily with Flask, and there is a dedicated extension, Flask‑MongoEngine, that simplifies configuration and adds extra features for web development.

Is there async support in MongoEngine?

MongoEngine was designed for synchronous use and does not provide built‑in async support. For asynchronous applications, consider using Beanie or Motor, which are built specifically for async/await patterns.

How do I test applications that use MongoEngine?

You can test MongoEngine applications using several approaches: mongomock for in‑memory MongoDB mocking, a separate test MongoDB instance, or Docker containers with MongoDB for isolated testing environments.

Why choose MongoEngine over PyMongo?

MongoEngine offers a higher level of abstraction compared to PyMongo. Its main advantages are declarative model definitions, automatic data validation, a user‑friendly query API, built‑in relationship handling, a signal system, and more readable code.

How can I optimize query performance in MongoEngine?

To optimize performance, create indexes on frequently queried fields, use projections to fetch only needed data, apply limit() and skip() for pagination, employ select_related() to pre‑fetch related documents, and analyze queries with explain().

Does MongoEngine support transactions?

Yes, MongoEngine supports transactions for MongoDB version 4.0 and higher. Transactions can be used via MongoDB sessions to ensure ACID properties across multiple operations.

Can MongoEngine work with existing MongoDB databases?

Yes. MongoEngine can operate with existing MongoDB databases. You simply define models that match the structure of existing documents and, if needed, specify collection names in the model metadata.

How do I handle data migrations when the schema changes in MongoEngine?

MongoDB does not enforce a rigid schema, so many changes can be applied without migrations. For complex changes, you can write migration scripts, use document update methods, or implement application‑level validation to handle legacy document structures.

Conclusion

MongoEngine is a powerful and convenient tool for working with MongoDB in Python applications. The library offers an intuitive API, automatic data validation, relationship handling, and many other useful features.

Key benefits of MongoEngine:

• Ease of use and Pythonic syntax
• Rich field type system with built‑in validation
• Support for indexes and query optimization
• Integration with popular web frameworks
• Active community and regular updates

MongoEngine is ideal for projects where rapid development, code readability, and reliable data handling are essential. It is successfully used in thousands of projects worldwide and continues to evolve, adapting to new MongoDB capabilities and developer needs.