tessl/pypi-mongomock

Fake pymongo stub for testing simple MongoDB-dependent code

—

Pending

Overview

Eval results

Files

Error Handling and Exceptions

Name: tessl/pypi-mongomock
Author: tessl

Complete exception hierarchy mirroring PyMongo's error system with support for operation failures, write errors, and configuration errors. Proper error handling ensures robust application behavior.

Capabilities

Base Exception Classes

Foundation exception classes that form the hierarchy for all MongoDB-related errors.

class PyMongoError(Exception):
    """
    Base exception class for all MongoDB-related errors.
    
    All mongomock exceptions inherit from this base class,
    providing consistent error handling patterns.
    """

class OperationFailure(PyMongoError):
    """
    Raised when a database operation fails.
    
    Common base class for database operation errors including
    write failures, query errors, and command execution problems.
    """
    
    def __init__(self, message, code=None, details=None):
        """
        Parameters:
        - message: str, error message
        - code: int, MongoDB error code
        - details: dict, additional error details
        """
        
    @property
    def code(self):
        """int: MongoDB error code"""
        
    @property
    def details(self):
        """dict: additional error details"""

class ConfigurationError(PyMongoError):
    """
    Raised for client configuration errors.
    
    Indicates problems with connection strings, client options,
    or other configuration-related issues.
    """

Usage Example:

import mongomock

try:
    collection = mongomock.MongoClient().db.collection
    collection.insert_one({'_id': 1, 'data': 'test'})
    collection.insert_one({'_id': 1, 'data': 'duplicate'})  # Will raise error
except mongomock.PyMongoError as e:
    print(f"MongoDB error: {e}")
except mongomock.OperationFailure as e:
    print(f"Operation failed: {e}, Code: {e.code}")

Write Operation Errors

Specific exceptions for write operation failures including constraint violations and bulk operation errors.

class WriteError(OperationFailure):
    """
    Raised when a write operation fails.
    
    Base class for write-specific errors including constraint
    violations, validation failures, and write conflicts.
    """

class DuplicateKeyError(WriteError):
    """
    Raised when a write operation violates a unique constraint.
    
    Occurs when attempting to insert or update documents with
    duplicate values for fields with unique indexes.
    """

class BulkWriteError(OperationFailure):
    """
    Raised when a bulk write operation fails.
    
    Contains details about which operations succeeded and
    which failed during bulk write execution.
    """
    
    def __init__(self, results):
        """
        Parameters:
        - results: dict, bulk operation results with error details
        """

Usage Example:

import mongomock

collection = mongomock.MongoClient().db.users

# Create unique index
collection.create_index('email', unique=True)

try:
    # First insert succeeds
    collection.insert_one({'email': 'user@example.com', 'name': 'Alice'})
    
    # Second insert with same email fails
    collection.insert_one({'email': 'user@example.com', 'name': 'Bob'})
    
except mongomock.DuplicateKeyError as e:
    print(f"Duplicate key error: {e}")
    print(f"Error code: {e.code}")

# Bulk write error handling
try:
    bulk_ops = [
        {'insertOne': {'document': {'email': 'new@example.com'}}},
        {'insertOne': {'document': {'email': 'user@example.com'}}},  # Duplicate
        {'insertOne': {'document': {'email': 'another@example.com'}}}
    ]
    collection.bulk_write(bulk_ops)
    
except mongomock.BulkWriteError as e:
    print(f"Bulk write failed: {e}")
    print(f"Error details: {e.details}")

Collection and Database Errors

Exceptions related to collection and database operations including invalid names and operations.

class CollectionInvalid(PyMongoError):
    """
    Raised when attempting invalid operations on collections.
    
    Examples include creating collections with invalid options
    or performing operations on non-existent collections.
    """

class InvalidName(PyMongoError):
    """
    Raised when database or collection names are invalid.
    
    MongoDB has specific naming restrictions for databases
    and collections that must be followed.
    """

class InvalidOperation(PyMongoError):
    """
    Raised when attempting invalid operations.
    
    Covers various invalid operation scenarios including
    inappropriate method calls or invalid parameter combinations.
    """

Usage Example:

import mongomock

client = mongomock.MongoClient()

try:
    # Invalid database name
    db = client['invalid/name']
    db.collection.insert_one({'test': 'data'})
    
except mongomock.InvalidName as e:
    print(f"Invalid name: {e}")

try:
    # Invalid collection operation
    db = client.testdb
    db.create_collection('existing_collection')
    db.create_collection('existing_collection')  # May raise CollectionInvalid
    
except mongomock.CollectionInvalid as e:
    print(f"Collection invalid: {e}")

try:
    # Invalid operation example
    collection = db.collection
    collection.find_one_and_update({}, {})  # Invalid empty update
    
except mongomock.InvalidOperation as e:
    print(f"Invalid operation: {e}")

URI and Configuration Errors

Exceptions for connection string parsing and configuration problems.

class InvalidURI(ConfigurationError):
    """
    Raised when MongoDB connection URI is malformed.
    
    Occurs when connection strings don't follow MongoDB URI
    format specifications or contain invalid parameters.
    """

Usage Example:

import mongomock

try:
    # Invalid URI format
    client = mongomock.MongoClient('invalid://bad-uri-format')
    
except mongomock.InvalidURI as e:
    print(f"Invalid URI: {e}")

try:
    # Configuration error
    client = mongomock.MongoClient(invalid_option='bad_value')
    
except mongomock.ConfigurationError as e:
    print(f"Configuration error: {e}")

Error Handling Patterns

Best practices for handling mongomock exceptions in applications.

Specific Exception Handling:

import mongomock

def safe_insert_user(collection, user_data):
    """Safely insert user with proper error handling."""
    try:
        result = collection.insert_one(user_data)
        return result.inserted_id
        
    except mongomock.DuplicateKeyError:
        print("User already exists with this email")
        return None
        
    except mongomock.WriteError as e:
        print(f"Write operation failed: {e}")
        return None
        
    except mongomock.OperationFailure as e:
        print(f"Database operation failed: {e.code} - {e}")
        return None

# Usage
collection = mongomock.MongoClient().db.users
collection.create_index('email', unique=True)

user_id = safe_insert_user(collection, {
    'email': 'user@example.com',
    'name': 'Alice'
})

Bulk Operation Error Handling:

def safe_bulk_insert(collection, documents):
    """Handle bulk insert with error recovery."""
    try:
        result = collection.insert_many(documents, ordered=False)
        return result.inserted_ids
        
    except mongomock.BulkWriteError as e:
        print(f"Bulk operation partially failed: {e}")
        
        # Access bulk write result details
        if hasattr(e, 'details'):
            write_errors = e.details.get('writeErrors', [])
            for error in write_errors:
                print(f"Error at index {error['index']}: {error['errmsg']}")
        
        # Return successful insertions
        return getattr(e, 'inserted_ids', [])
        
    except mongomock.OperationFailure as e:
        print(f"Bulk operation completely failed: {e}")
        return []

# Usage with error recovery
documents = [
    {'email': 'user1@example.com', 'name': 'User 1'},
    {'email': 'user2@example.com', 'name': 'User 2'},
    {'email': 'user1@example.com', 'name': 'Duplicate'},  # Will fail
    {'email': 'user3@example.com', 'name': 'User 3'}
]

inserted_ids = safe_bulk_insert(collection, documents)
print(f"Successfully inserted {len(inserted_ids)} documents")

Generic Error Handling:

def robust_query(collection, query_filter):
    """Perform query with comprehensive error handling."""
    try:
        return list(collection.find(query_filter))
        
    except mongomock.OperationFailure as e:
        print(f"Query failed: {e}")
        
        # Check specific error codes
        if e.code == 2:  # BadValue
            print("Invalid query parameters")
        elif e.code == 11000:  # DuplicateKey
            print("Duplicate key error during query")
        else:
            print(f"Unknown operation failure: {e.code}")
            
        return []
        
    except mongomock.PyMongoError as e:
        print(f"General MongoDB error: {e}")
        return []
        
    except Exception as e:
        print(f"Unexpected error: {e}")
        return []

Error Code Constants

Common MongoDB error codes that may be encountered.

# Common MongoDB error codes
ERROR_CODES = {
    2: 'BadValue',           # Invalid parameter value
    11000: 'DuplicateKey',   # Duplicate key error
    11001: 'DuplicateKey',   # Duplicate key on update
    12582: 'DuplicateKey',   # Duplicate key in capped collection
    16550: 'CommandNotFound', # Unknown command
    26: 'NamespaceNotFound', # Collection/database not found
    50: 'MaxTimeMSExpired',  # Operation exceeded time limit
    96: 'OperationFailed',   # Generic operation failure
}

Usage Example:

import mongomock

def handle_error_by_code(error):
    """Handle errors based on MongoDB error codes."""
    if hasattr(error, 'code'):
        if error.code == 11000:
            return "Duplicate key - record already exists"
        elif error.code == 2:
            return "Invalid parameter value provided"
        elif error.code == 26:
            return "Collection or database not found"
        else:
            return f"MongoDB error {error.code}: {error}"
    return str(error)

# Example usage
try:
    collection = mongomock.MongoClient().db.test
    collection.create_index('email', unique=True)
    collection.insert_one({'email': 'test@example.com'})
    collection.insert_one({'email': 'test@example.com'})  # Duplicate
    
except mongomock.PyMongoError as e:
    error_message = handle_error_by_code(e)
    print(f"Handled error: {error_message}")

Testing Error Conditions

Patterns for testing error handling in applications using mongomock.

import unittest
import mongomock

class TestErrorHandling(unittest.TestCase):
    def setUp(self):
        self.client = mongomock.MongoClient()
        self.collection = self.client.db.test
        self.collection.create_index('email', unique=True)
    
    def test_duplicate_key_error(self):
        """Test handling of duplicate key errors."""
        self.collection.insert_one({'email': 'test@example.com'})
        
        with self.assertRaises(mongomock.DuplicateKeyError):
            self.collection.insert_one({'email': 'test@example.com'})
    
    def test_invalid_operation_error(self):
        """Test handling of invalid operations."""
        with self.assertRaises(mongomock.InvalidOperation):
            self.collection.update_one({}, {})  # Empty update
    
    def test_collection_invalid_error(self):
        """Test collection validation errors."""
        with self.assertRaises(mongomock.CollectionInvalid):
            self.client.db.create_collection('invalid..name')
    
    def test_error_recovery(self):
        """Test error recovery patterns."""
        try:
            self.collection.insert_one({'email': 'test@example.com'})
            self.collection.insert_one({'email': 'test@example.com'})
        except mongomock.DuplicateKeyError:
            # Recover by updating instead
            result = self.collection.update_one(
                {'email': 'test@example.com'},
                {'$set': {'updated': True}}
            )
            self.assertEqual(result.modified_count, 1)

Install with Tessl CLI