Exceptions¶

QuantumEngine provides a comprehensive exception hierarchy for handling errors across different database backends and operations.

Overview ¶

QuantumEngine’s exception system provides:

Consistent error handling across different backends
Specific exception types for different error categories
Detailed error information for debugging and logging
Backend-specific exceptions for specialized error handling
Validation exceptions for field and document validation

Exception Hierarchy ¶

All QuantumEngine exceptions inherit from the base QuantumEngineException class:

QuantumEngineException
├── ValidationError
│   ├── FieldValidationError
│   └── DocumentValidationError
├── DatabaseError
│   ├── ConnectionError
│   ├── QueryError
│   └── TransactionError
├── BackendError
│   ├── SurrealDBError
│   └── ClickHouseError
├── DocumentError
│   ├── DoesNotExist
│   ├── MultipleObjectsReturned
│   └── DocumentIntegrityError
└── ConfigurationError
    ├── BackendConfigurationError
    └── FieldConfigurationError

Core Exceptions ¶

Base Exception ¶

from quantumengine.exceptions import QuantumEngineException

try:
    # QuantumEngine operation
    pass
except QuantumEngineException as e:
    print(f"QuantumEngine error: {e}")

ValidationError ¶

Raised when data validation fails:

from quantumengine.exceptions import ValidationError
from quantumengine import Document
from quantumengine.fields import StringField, IntField

class User(Document):
    name = StringField(required=True)
    age = IntField(min_value=0, max_value=150)

try:
    user = User(name="", age=-5)  # Invalid data
    user.save()
except ValidationError as e:
    print(f"Validation failed: {e}")
    print(f"Field errors: {e.field_errors}")

DatabaseError ¶

Raised for database-related errors:

from quantumengine.exceptions import DatabaseError

try:
    users = User.objects.all()
except DatabaseError as e:
    print(f"Database error: {e}")
    print(f"Error code: {e.code}")
    print(f"Backend: {e.backend}")

Field Validation Exceptions ¶

FieldValidationError ¶

Raised when individual field validation fails:

from quantumengine.exceptions import FieldValidationError
from quantumengine.fields import EmailField

try:
    email_field = EmailField()
    email_field.validate("invalid-email")
except FieldValidationError as e:
    print(f"Field validation error: {e}")
    print(f"Field name: {e.field_name}")
    print(f"Invalid value: {e.value}")

Common field validation errors:

from quantumengine.fields import StringField, IntField, EmailField

# Required field validation
try:
    name_field = StringField(required=True)
    name_field.validate(None)
except FieldValidationError as e:
    print("Required field is missing")

# Range validation
try:
    age_field = IntField(min_value=0, max_value=150)
    age_field.validate(200)
except FieldValidationError as e:
    print("Value out of range")

# Format validation
try:
    email_field = EmailField()
    email_field.validate("not-an-email")
except FieldValidationError as e:
    print("Invalid email format")

Document Validation Exceptions ¶

DocumentValidationError ¶

Raised when document-level validation fails:

from quantumengine.exceptions import DocumentValidationError

class User(Document):
    username = StringField(required=True, unique=True)
    email = EmailField(required=True)

    def clean(self):
        # Custom validation
        if self.username.lower() == 'admin':
            raise DocumentValidationError("Username 'admin' is reserved")

try:
    user = User(username='admin', email='admin@example.com')
    user.save()
except DocumentValidationError as e:
    print(f"Document validation error: {e}")

Database Operation Exceptions ¶

DoesNotExist ¶

Raised when querying for a non-existent object:

from quantumengine.exceptions import DoesNotExist

try:
    user = User.objects.get(username='nonexistent')
except DoesNotExist:
    print("User does not exist")

# Each document class has its own DoesNotExist exception
try:
    user = User.objects.get(username='nonexistent')
except User.DoesNotExist:
    print("User does not exist")

MultipleObjectsReturned ¶

Raised when get() returns multiple objects:

from quantumengine.exceptions import MultipleObjectsReturned

try:
    user = User.objects.get(age=25)  # Multiple users with age 25
except MultipleObjectsReturned:
    print("Multiple users found")

# Safe alternative
user = User.objects.filter(age=25).first()

Connection Exceptions ¶

ConnectionError ¶

Raised when database connection fails:

from quantumengine.exceptions import ConnectionError

try:
    User.objects.all()
except ConnectionError as e:
    print(f"Connection failed: {e}")
    print(f"Backend: {e.backend}")
    print(f"Connection details: {e.connection_info}")

Query Exceptions ¶

QueryError ¶

Raised when query execution fails:

from quantumengine.exceptions import QueryError

try:
    # Invalid query
    User.objects.filter(invalid_field='value')
except QueryError as e:
    print(f"Query error: {e}")
    print(f"Query: {e.query}")
    print(f"Parameters: {e.params}")

Backend-Specific Exceptions ¶

SurrealDBError ¶

Specific to SurrealDB backend operations:

from quantumengine.exceptions import SurrealDBError

try:
    # SurrealDB-specific operation
    User.objects.raw("INVALID SURREALQL")
except SurrealDBError as e:
    print(f"SurrealDB error: {e}")
    print(f"Error code: {e.surrealdb_code}")
    print(f"Details: {e.details}")

ClickHouseError ¶

Specific to ClickHouse backend operations:

from quantumengine.exceptions import ClickHouseError

try:
    # ClickHouse-specific operation
    Analytics.objects.raw("INVALID SQL")
except ClickHouseError as e:
    print(f"ClickHouse error: {e}")
    print(f"Error code: {e.clickhouse_code}")
    print(f"Query: {e.query}")

Configuration Exceptions ¶

BackendConfigurationError ¶

Raised when backend configuration is invalid:

from quantumengine.exceptions import BackendConfigurationError
from quantumengine import configure_backend

try:
    configure_backend('surrealdb', {
        'url': 'invalid-url',
        'namespace': '',  # Invalid empty namespace
    })
except BackendConfigurationError as e:
    print(f"Configuration error: {e}")
    print(f"Backend: {e.backend}")
    print(f"Invalid config: {e.config}")

Transaction Exceptions ¶

TransactionError ¶

Raised when transaction operations fail:

from quantumengine.exceptions import TransactionError
from quantumengine import transaction

try:
    with transaction.atomic():
        user = User.objects.create(name='John')
        # Simulate error
        raise Exception("Something went wrong")
except TransactionError as e:
    print(f"Transaction error: {e}")
    print(f"Transaction ID: {e.transaction_id}")

Error Handling Best Practices ¶

Specific Exception Handling ¶

Catch specific exceptions rather than generic ones:

from quantumengine.exceptions import DoesNotExist, ValidationError

def get_user_safely(username):
    try:
        return User.objects.get(username=username)
    except DoesNotExist:
        return None
    except ValidationError as e:
        logger.error(f"Validation error for username {username}: {e}")
        return None

Logging Exceptions ¶

Log exceptions with context for debugging:

import logging
from quantumengine.exceptions import DatabaseError

logger = logging.getLogger(__name__)

def handle_database_error(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except DatabaseError as e:
            logger.error(f"Database error in {func.__name__}: {e}",
                       extra={
                           'backend': e.backend,
                           'error_code': e.code,
                           'function': func.__name__,
                           'args': args,
                           'kwargs': kwargs
                       })
            raise
    return wrapper

@handle_database_error
def get_user_by_id(user_id):
    return User.objects.get(id=user_id)

Graceful Degradation ¶

Handle exceptions gracefully in production:

from quantumengine.exceptions import ConnectionError, QueryError

def get_user_with_fallback(username):
    try:
        return User.objects.get(username=username)
    except ConnectionError:
        # Fallback to cache or return default
        return get_user_from_cache(username)
    except QueryError:
        # Log error and return safe default
        logger.warning(f"Query error for username: {username}")
        return create_anonymous_user()

Custom Exceptions ¶

Create application-specific exceptions:

from quantumengine.exceptions import QuantumEngineException

class UserNotActiveError(QuantumEngineException):
    """Raised when trying to perform operations on inactive user"""
    pass

class InsufficientPermissionsError(QuantumEngineException):
    """Raised when user lacks required permissions"""
    pass

def activate_user(user_id):
    try:
        user = User.objects.get(id=user_id)
        if not user.is_active:
            raise UserNotActiveError(f"User {user_id} is not active")
        return user
    except DoesNotExist:
        raise UserNotActiveError(f"User {user_id} does not exist")

Exception Context ¶

Exceptions provide context for debugging:

from quantumengine.exceptions import QueryError

try:
    User.objects.filter(age__invalid_lookup=25)
except QueryError as e:
    print(f"Error: {e}")
    print(f"Query: {e.query}")
    print(f"Backend: {e.backend}")
    print(f"Traceback: {e.traceback}")

Testing Exceptions ¶

Test exception handling in your code:

import pytest
from quantumengine.exceptions import DoesNotExist, ValidationError

def test_user_not_found():
    with pytest.raises(DoesNotExist):
        User.objects.get(username='nonexistent')

def test_validation_error():
    with pytest.raises(ValidationError) as exc_info:
        user = User(name='', age=-5)
        user.save()

    assert 'name' in exc_info.value.field_errors
    assert 'age' in exc_info.value.field_errors

Exception Middleware ¶

Handle exceptions at the application level:

from quantumengine.exceptions import QuantumEngineException

class QuantumEngineExceptionMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        try:
            response = self.get_response(request)
            return response
        except QuantumEngineException as e:
            # Log and handle QuantumEngine exceptions
            logger.error(f"QuantumEngine error: {e}")
            return handle_quantumengine_error(e)