tessl/pypi-mongomock

Fake pymongo stub for testing simple MongoDB-dependent code

—

Pending

Overview

Eval results

Files

Collection CRUD Operations

Name: tessl/pypi-mongomock
Author: tessl

Complete Create, Read, Update, Delete operations with support for bulk operations, complex queries, and MongoDB-compatible filtering and sorting. The Collection class provides the primary interface for document manipulation.

Capabilities

Document Insertion

Insert single or multiple documents into collections with support for validation bypass and bulk operations.

def insert_one(self, document, bypass_document_validation=False, session=None):
    """
    Insert a single document.

    Parameters:
    - document: dict, document to insert
    - bypass_document_validation: bool, skip document validation
    - session: ClientSession, session to use (ignored)

    Returns:
    InsertOneResult: result with inserted_id and acknowledged status

    Raises:
    DuplicateKeyError: if duplicate key constraint violated
    WriteError: if write operation fails
    """

def insert_many(self, documents, ordered=True, 
               bypass_document_validation=False, session=None):
    """
    Insert multiple documents.

    Parameters:
    - documents: list of dict, documents to insert
    - ordered: bool, stop on first error if True
    - bypass_document_validation: bool, skip document validation
    - session: ClientSession, session to use (ignored)

    Returns:
    InsertManyResult: result with inserted_ids and acknowledged status

    Raises:
    BulkWriteError: if bulk operation fails
    DuplicateKeyError: if duplicate key constraint violated
    """

Usage Example:

collection = mongomock.MongoClient().db.users

# Insert single document
result = collection.insert_one({
    'name': 'Alice',
    'email': 'alice@example.com',
    'age': 30
})
print(f"Inserted ID: {result.inserted_id}")

# Insert multiple documents
users = [
    {'name': 'Bob', 'age': 25},
    {'name': 'Charlie', 'age': 35},
    {'name': 'David', 'age': 28}
]
result = collection.insert_many(users)
print(f"Inserted {len(result.inserted_ids)} documents")

Document Querying

Find documents using MongoDB query syntax with support for projection, sorting, limiting, and cursor operations.

def find(self, filter=None, projection=None, skip=0, limit=0, 
         sort=None, cursor_type=None, collation=None, hint=None, 
         session=None, **kwargs):
    """
    Find documents matching filter criteria.

    Parameters:
    - filter: dict, query filter (None matches all)
    - projection: dict or list, fields to include/exclude
    - skip: int, number of documents to skip
    - limit: int, maximum number of documents to return
    - sort: list of (key, direction) tuples or dict
    - cursor_type: CursorType, cursor type (ignored)
    - collation: dict, collation specification (limited support)
    - hint: str or list, index hint
    - session: ClientSession, session to use (ignored)
    - **kwargs: additional query options

    Returns:
    Cursor: cursor over matching documents
    """

def find_one(self, filter=None, *args, **kwargs):
    """
    Find a single document.

    Parameters:
    - filter: dict, query filter (None matches any)
    - *args, **kwargs: same as find()

    Returns:
    dict or None: matching document or None if not found
    """

Usage Example:

collection = mongomock.MongoClient().db.users

# Find all documents
all_users = list(collection.find())

# Find with filter
young_users = list(collection.find({'age': {'$lt': 30}}))

# Find with projection
names_only = list(collection.find({}, {'name': 1, '_id': 0}))

# Find with sorting and limiting
top_users = list(collection.find().sort('age', -1).limit(5))

# Find one document
user = collection.find_one({'name': 'Alice'})

# Complex query
adults = list(collection.find({
    'age': {'$gte': 18, '$lt': 65},
    'status': {'$in': ['active', 'pending']}
}))

Find and Modify Operations

Atomic find-and-modify operations that combine query and modification in single operations.

def find_one_and_delete(self, filter, projection=None, sort=None, **kwargs):
    """
    Find a document and delete it atomically.

    Parameters:
    - filter: dict, query filter
    - projection: dict or list, fields to return
    - sort: list of (key, direction) tuples
    - **kwargs: additional options

    Returns:
    dict or None: deleted document or None if not found
    """

def find_one_and_replace(self, filter, replacement, projection=None, 
                        sort=None, upsert=False, 
                        return_document=ReturnDocument.BEFORE, **kwargs):
    """
    Find a document and replace it atomically.

    Parameters:
    - filter: dict, query filter
    - replacement: dict, replacement document
    - projection: dict or list, fields to return
    - sort: list of (key, direction) tuples
    - upsert: bool, insert if no match found
    - return_document: ReturnDocument.BEFORE or AFTER
    - **kwargs: additional options

    Returns:
    dict or None: original or modified document
    """

def find_one_and_update(self, filter, update, projection=None, sort=None, 
                       upsert=False, return_document=ReturnDocument.BEFORE, 
                       **kwargs):
    """
    Find a document and update it atomically.

    Parameters:
    - filter: dict, query filter  
    - update: dict, update operations
    - projection: dict or list, fields to return
    - sort: list of (key, direction) tuples
    - upsert: bool, insert if no match found
    - return_document: ReturnDocument.BEFORE or AFTER
    - **kwargs: additional options

    Returns:
    dict or None: original or modified document
    """

Usage Example:

from mongomock import ReturnDocument

collection = mongomock.MongoClient().db.users

# Find and delete
deleted_user = collection.find_one_and_delete({'name': 'Bob'})

# Find and replace
new_user = collection.find_one_and_replace(
    {'name': 'Alice'},
    {'name': 'Alice Smith', 'age': 31, 'email': 'alice.smith@example.com'},
    return_document=ReturnDocument.AFTER
)

# Find and update  
updated_user = collection.find_one_and_update(
    {'name': 'Charlie'},
    {'$set': {'age': 36}, '$inc': {'login_count': 1}},
    return_document=ReturnDocument.AFTER
)

Document Updates

Update documents using MongoDB update operators with support for single and multiple document updates.

def update_one(self, filter, update, upsert=False, 
              bypass_document_validation=False, collation=None, 
              array_filters=None, hint=None, session=None):
    """
    Update a single document.

    Parameters:
    - filter: dict, query filter
    - update: dict, update operations
    - upsert: bool, insert if no match found
    - bypass_document_validation: bool, skip validation
    - collation: dict, collation specification (limited support)
    - array_filters: list, array element filters
    - hint: str or list, index hint
    - session: ClientSession, session to use (ignored)

    Returns:
    UpdateResult: result with match/modification counts

    Raises:
    WriteError: if update operation fails
    """

def update_many(self, filter, update, upsert=False, 
               bypass_document_validation=False, collation=None, 
               array_filters=None, hint=None, session=None):
    """
    Update multiple documents.

    Parameters:
    - filter: dict, query filter
    - update: dict, update operations  
    - upsert: bool, insert if no match found
    - bypass_document_validation: bool, skip validation
    - collation: dict, collation specification (limited support)
    - array_filters: list, array element filters
    - hint: str or list, index hint
    - session: ClientSession, session to use (ignored)

    Returns:
    UpdateResult: result with match/modification counts

    Raises:
    WriteError: if update operation fails
    """

def replace_one(self, filter, replacement, upsert=False, 
               bypass_document_validation=False, collation=None, 
               hint=None, session=None):
    """
    Replace a single document.

    Parameters:
    - filter: dict, query filter
    - replacement: dict, replacement document
    - upsert: bool, insert if no match found
    - bypass_document_validation: bool, skip validation
    - collation: dict, collation specification (limited support)
    - hint: str or list, index hint
    - session: ClientSession, session to use (ignored)

    Returns:
    UpdateResult: result with match/modification counts

    Raises:
    WriteError: if replace operation fails
    """

Usage Example:

collection = mongomock.MongoClient().db.users

# Update single document
result = collection.update_one(
    {'name': 'Alice'},
    {'$set': {'age': 31}, '$inc': {'login_count': 1}}
)
print(f"Modified: {result.modified_count}")

# Update multiple documents
result = collection.update_many(
    {'age': {'$lt': 30}},
    {'$set': {'category': 'young'}}
)
print(f"Modified: {result.modified_count}")

# Replace document
result = collection.replace_one(
    {'name': 'Bob'},
    {'name': 'Robert', 'age': 26, 'email': 'robert@example.com'}
)

# Upsert (update or insert)
result = collection.update_one(
    {'name': 'Eve'},
    {'$set': {'age': 25}},
    upsert=True
)
if result.upserted_id:
    print(f"Inserted new document: {result.upserted_id}")

Document Deletion

Remove documents from collections with support for single and multiple document deletion.

def delete_one(self, filter, collation=None, hint=None, session=None):
    """
    Delete a single document.

    Parameters:
    - filter: dict, query filter
    - collation: dict, collation specification (limited support)
    - hint: str or list, index hint
    - session: ClientSession, session to use (ignored)

    Returns:
    DeleteResult: result with deletion count

    Raises:
    WriteError: if delete operation fails
    """

def delete_many(self, filter, collation=None, hint=None, session=None):
    """
    Delete multiple documents.

    Parameters:
    - filter: dict, query filter
    - collation: dict, collation specification (limited support)
    - hint: str or list, index hint
    - session: ClientSession, session to use (ignored)

    Returns:
    DeleteResult: result with deletion count

    Raises:
    WriteError: if delete operation fails
    """

Usage Example:

collection = mongomock.MongoClient().db.users

# Delete single document
result = collection.delete_one({'name': 'Alice'})
print(f"Deleted: {result.deleted_count}")

# Delete multiple documents
result = collection.delete_many({'age': {'$gt': 65}})
print(f"Deleted: {result.deleted_count}")

# Delete all documents
result = collection.delete_many({})
print(f"Deleted all {result.deleted_count} documents")

Document Counting

Count documents in collections with support for filtering and estimation.

def count_documents(self, filter, **kwargs):
    """
    Count documents matching filter.

    Parameters:
    - filter: dict, query filter
    - **kwargs: additional count options

    Returns:
    int: number of matching documents
    """

def estimated_document_count(self, **kwargs):
    """
    Estimate total document count in collection.

    Parameters:
    - **kwargs: additional options (ignored)

    Returns:
    int: estimated document count
    """

Usage Example:

collection = mongomock.MongoClient().db.users

# Count with filter
young_count = collection.count_documents({'age': {'$lt': 30}})
print(f"Young users: {young_count}")

# Count all documents
total_count = collection.count_documents({})

# Estimated count (same as exact count in mongomock)
estimated = collection.estimated_document_count()

Collection Constructor

class Collection:
    def __init__(self, database, name, create=False, **kwargs):
        """
        Create a Collection instance.
        
        Note: Collections are typically created via Database.get_collection()
        rather than direct instantiation.

        Parameters:
        - database: Database, parent database instance
        - name: str, collection name
        - create: bool, explicitly create collection
        - **kwargs: collection options
        """

Install with Tessl CLI