Fake pymongo stub for testing simple MongoDB-dependent code
npx @tessl/cli install tessl/pypi-mongomock@4.3.0A comprehensive MongoDB mocking library that provides a complete fake implementation of PyMongo's API for testing Python applications. Mongomock enables developers to test MongoDB-dependent code without requiring an actual MongoDB instance, offering full compatibility with PyMongo interfaces including collections, databases, queries, aggregation pipelines, and GridFS functionality.
pip install mongomockimport mongomockMain classes and functions:
from mongomock import MongoClient, Database, Collection, ObjectId
from mongomock import patch, ignore_feature, warn_on_featureError classes:
from mongomock import (
OperationFailure, DuplicateKeyError,
CollectionInvalid, InvalidName
)Configuration classes:
from mongomock import WriteConcernimport mongomock
# Create a mock MongoDB client
client = mongomock.MongoClient()
# Get a database
db = client.test_database
# Get a collection
collection = db.test_collection
# Insert documents
collection.insert_one({"name": "Alice", "age": 30})
collection.insert_many([
{"name": "Bob", "age": 25},
{"name": "Charlie", "age": 35}
])
# Query documents
user = collection.find_one({"name": "Alice"})
all_users = list(collection.find({"age": {"$gte": 25}}))
# Update documents
collection.update_one({"name": "Alice"}, {"$set": {"age": 31}})
collection.update_many({"age": {"$lt": 30}}, {"$set": {"status": "young"}})
# Delete documents
collection.delete_one({"name": "Bob"})
collection.delete_many({"age": {"$gt": 40}})
# Count documents
count = collection.count_documents({"age": {"$gte": 25}})Mongomock mirrors PyMongo's architecture while providing in-memory storage and mock behavior:
This architecture ensures drop-in compatibility with PyMongo while providing predictable, controllable behavior for testing scenarios.
Core client functionality for establishing mock connections and managing database access with configuration options and server information.
class MongoClient:
def __init__(self, host=None, port=None, document_class=dict,
tz_aware=False, connect=True, **kwargs): ...
def get_database(self, name=None, **kwargs): ...
def list_database_names(self): ...
def server_info(self): ...
def close(self): ...Client and Connection Management
Database-level operations including collection management, database commands, and configuration with support for read/write concerns and codec options.
class Database:
def get_collection(self, name, **kwargs): ...
def list_collection_names(self, filter=None, session=None): ...
def create_collection(self, name, **kwargs): ...
def drop_collection(self, name_or_collection, session=None): ...
def command(self, command, **kwargs): ...Complete Create, Read, Update, Delete operations with support for bulk operations, complex queries, and MongoDB-compatible filtering and sorting.
class Collection:
def insert_one(self, document, **kwargs): ...
def insert_many(self, documents, **kwargs): ...
def find(self, filter=None, **kwargs): ...
def find_one(self, filter=None, **kwargs): ...
def update_one(self, filter, update, **kwargs): ...
def update_many(self, filter, update, **kwargs): ...
def delete_one(self, filter, **kwargs): ...
def delete_many(self, filter, **kwargs): ...Index management operations including creation, deletion, and inspection of indexes with support for compound indexes, text indexes, and index options.
def create_index(self, key_or_list, **kwargs): ...
def create_indexes(self, indexes, session=None): ...
def drop_index(self, index_or_name, session=None): ...
def list_indexes(self, session=None): ...
def index_information(self, session=None): ...Aggregation pipeline operations and data analysis functions including pipeline stages, aggregation operators, and distinct value queries.
def aggregate(self, pipeline, session=None, **kwargs): ...
def distinct(self, key, filter=None, session=None): ...
def count_documents(self, filter, **kwargs): ...
def estimated_document_count(self, **kwargs): ...Cursor management for query results with support for sorting, limiting, skipping, and batch processing of result sets.
class Cursor:
def sort(self, key_or_list, direction=None): ...
def skip(self, count): ...
def limit(self, count): ...
def batch_size(self, count): ...
def hint(self, hint): ...Configuration classes for controlling read/write behavior, codec options, and database connection parameters.
class WriteConcern:
def __init__(self, w=None, wtimeout=None, j=None, fsync=None): ...
class ReadConcern:
def __init__(self, level=None): ...
class CodecOptions:
def __init__(self, document_class=dict, tz_aware=False, **kwargs): ...Complete exception hierarchy mirroring PyMongo's error system with support for operation failures, write errors, and configuration errors.
class OperationFailure(PyMongoError): ...
class WriteError(OperationFailure): ...
class DuplicateKeyError(WriteError): ...
class BulkWriteError(OperationFailure): ...
class CollectionInvalid(PyMongoError): ...Utilities for test integration including patching, feature management, and development helpers for seamless testing workflows.
def patch(servers='localhost', on_new='error'): ...
def ignore_feature(feature): ...
def warn_on_feature(feature): ...
def enable_gridfs_integration(): ...Testing and Integration Utilities
from typing import Any, Dict, List, Optional, Union
# Document types
Document = Dict[str, Any]
Filter = Dict[str, Any]
Update = Dict[str, Any]
Projection = Union[Dict[str, Any], List[str]]
# Sort specification
SortSpec = Union[str, List[tuple]]
# Index specification
IndexSpec = Union[str, List[tuple]]
# Session type (placeholder for compatibility)
ClientSession = Any
# Result document
ResultDocument = Dict[str, Any]