tessl/pypi-mongomock

Fake pymongo stub for testing simple MongoDB-dependent code

—

Pending

Overview

Eval results

Files

Cursors and Result Iteration

Name: tessl/pypi-mongomock
Author: tessl

Cursor management for query results with support for sorting, limiting, skipping, and batch processing of result sets. Cursors provide efficient iteration over query results with chainable operations.

Capabilities

Cursor Creation and Iteration

Cursors are created by find operations and provide iterator interface for accessing results.

class Cursor:
    def __iter__(self):
        """
        Iterator interface for cursor.

        Returns:
        Iterator[dict]: iterator over documents
        """

    def __next__(self):
        """
        Get next document.

        Returns:
        dict: next document

        Raises:
        StopIteration: when no more documents
        """

    def next(self):
        """
        Get next document (Python 2 compatibility).

        Returns:
        dict: next document

        Raises:
        StopIteration: when no more documents
        """

Usage Example:

collection = mongomock.MongoClient().db.users

# Get cursor from find operation
cursor = collection.find({'age': {'$gte': 18}})

# Iterate over results
for document in cursor:
    print(f"User: {document['name']}, Age: {document['age']}")

# Convert to list
users = list(collection.find({'status': 'active'}))

Result Ordering

Sort query results using single or multiple sort criteria with ascending or descending order.

def sort(self, key_or_list, direction=None):
    """
    Sort the cursor results.

    Parameters:
    - key_or_list: str or list, sort key or list of (key, direction) tuples
    - direction: int, sort direction (1 for ascending, -1 for descending)

    Returns:
    Cursor: cursor with sorting applied
    """

Usage Example:

from mongomock import ASCENDING, DESCENDING

collection = mongomock.MongoClient().db.users

# Sort by single field
cursor = collection.find().sort('age', ASCENDING)
cursor = collection.find().sort('name', DESCENDING)

# Sort by multiple fields
cursor = collection.find().sort([
    ('age', DESCENDING),
    ('name', ASCENDING)
])

# Alternative syntax
cursor = collection.find().sort('age', 1)  # Ascending
cursor = collection.find().sort('age', -1) # Descending

# Chain sorting with other operations
results = list(
    collection.find({'status': 'active'})
    .sort('created_at', -1)
    .limit(10)
)

Result Limiting and Skipping

Control the number of results returned and implement pagination through skip and limit operations.

def skip(self, count):
    """
    Skip the specified number of documents.

    Parameters:
    - count: int, number of documents to skip

    Returns:
    Cursor: cursor with skip applied
    """

def limit(self, count):
    """
    Limit the number of returned documents.

    Parameters:
    - count: int, maximum number of documents to return

    Returns:
    Cursor: cursor with limit applied
    """

Usage Example:

collection = mongomock.MongoClient().db.users

# Get first 10 results
first_page = list(collection.find().limit(10))

# Skip first 10, get next 10 (pagination)
second_page = list(collection.find().skip(10).limit(10))

# Get top 5 oldest users
oldest_users = list(
    collection.find()
    .sort('birth_date', 1)
    .limit(5)
)

# Complex pagination
page_size = 20
page_number = 3
page_results = list(
    collection.find({'status': 'active'})
    .sort('created_at', -1)
    .skip((page_number - 1) * page_size)
    .limit(page_size)
)

Batch Processing

Configure batch size for efficient memory usage when processing large result sets.

def batch_size(self, count):
    """
    Set the cursor batch size.

    Parameters:
    - count: int, number of documents per batch

    Returns:
    Cursor: cursor with batch size configured
    """

Usage Example:

collection = mongomock.MongoClient().db.large_collection

# Process large dataset with controlled batch size
cursor = collection.find().batch_size(100)

for document in cursor:
    # Process each document
    process_document(document)

Query Hints and Optimization

Provide hints to query planner and control query execution behavior.

def hint(self, hint):
    """
    Provide a hint to the query planner.

    Parameters:
    - hint: str or list, index hint specification

    Returns:
    Cursor: cursor with hint applied
    """

def max_time_ms(self, max_time_ms):
    """
    Set maximum execution time for the query.

    Parameters:
    - max_time_ms: int, maximum execution time in milliseconds

    Returns:
    Cursor: cursor with time limit applied
    """

def allow_disk_use(self, allow_disk_use=False):
    """
    Allow query to use disk for sorting large result sets.

    Parameters:
    - allow_disk_use: bool, whether to allow disk usage

    Returns:
    Cursor: cursor with disk usage setting applied
    """

Usage Example:

collection = mongomock.MongoClient().db.users

# Use index hint
cursor = collection.find({'email': 'user@example.com'}).hint('email_1')

# Set query timeout
cursor = collection.find().max_time_ms(5000)  # 5 second timeout

# Allow disk usage for large sorts
cursor = collection.find().sort('created_at', -1).allow_disk_use(True)

# Combine optimization options
optimized_cursor = (
    collection.find({'status': 'active'})
    .hint('status_1_created_at_-1')
    .max_time_ms(10000)
    .batch_size(50)
)

Cursor Management

Control cursor lifecycle and state management operations.

def close(self):
    """
    Close the cursor and free resources.

    Returns:
    None
    """

def clone(self):
    """
    Create a copy of the cursor.

    Returns:
    Cursor: cloned cursor instance
    """

def rewind(self):
    """
    Rewind cursor to beginning.

    Returns:
    Cursor: rewound cursor
    """

def collation(self, collation=None):
    """
    Set collation for the cursor.

    Parameters:
    - collation: dict, collation specification

    Returns:
    Cursor: cursor with collation applied
    """

Usage Example:

collection = mongomock.MongoClient().db.users

# Create cursor
cursor = collection.find({'status': 'active'})

# Clone cursor for reuse
cursor_copy = cursor.clone()

# Process first set of results
first_results = list(cursor.limit(10))

# Rewind and process again
cursor.rewind()
second_results = list(cursor.limit(5))

# Close cursor when done
cursor.close()

# Use collation for text sorting
text_cursor = collection.find().collation({
    'locale': 'en_US',
    'strength': 2  # Case insensitive
})

Cursor Properties

Access cursor state and configuration information.

@property
def alive(self):
    """
    bool: whether cursor is still active and has more results
    """

@property
def cursor_id(self):
    """
    int: unique cursor identifier
    """

@property
def address(self):
    """
    tuple: server address (always None in mongomock)
    """

Usage Example:

collection = mongomock.MongoClient().db.users
cursor = collection.find()

# Check if cursor has more results
if cursor.alive:
    next_doc = next(cursor)

print(f"Cursor ID: {cursor.cursor_id}")
print(f"Server address: {cursor.address}")

Distinct Values

Get distinct values for a field across the cursor results.

def distinct(self, key, session=None):
    """
    Get distinct values for a field.

    Parameters:
    - key: str, field name to get distinct values for
    - session: ClientSession, session to use (ignored)

    Returns:
    list: list of distinct values
    """

Usage Example:

collection = mongomock.MongoClient().db.users

# Get distinct values from cursor
cursor = collection.find({'status': 'active'})
distinct_ages = cursor.distinct('age')
distinct_departments = cursor.distinct('department')

print(f"Distinct ages: {distinct_ages}")
print(f"Distinct departments: {distinct_departments}")

Constants

# Sort directions (from mongomock.helpers)
ASCENDING = 1
DESCENDING = -1

# Cursor types
class CursorType:
    NON_TAILABLE = 0
    TAILABLE = 2
    TAILABLE_AWAIT = 130
    EXHAUST = 64

Install with Tessl CLI