tessl/pypi-python-arango

Python Driver for ArangoDB, a scalable multi-model database that natively supports documents, graphs, and search.

—

Pending

Overview

Eval results

Files

Error Handling & Types

Name: tessl/pypi-python-arango
Author: tessl

Comprehensive exception hierarchy and type definitions for structured error handling and type safety in python-arango operations. Includes operation-specific exceptions, error codes, and result wrappers.

Capabilities

Exception Hierarchy

Complete exception classes for all ArangoDB operations with inheritance from base exception types.

class ArangoError(Exception):
    """Base exception for all ArangoDB errors."""
    
    @property
    def error_code(self) -> int:
        """ArangoDB error code."""
        
    @property
    def error_message(self) -> str:
        """Error message."""
        
    @property
    def http_code(self) -> int:
        """HTTP status code."""

class ArangoServerError(ArangoError):
    """Server-side errors from ArangoDB."""

class ArangoClientError(ArangoError):
    """Client-side errors in python-arango."""

class AsyncExecuteError(ArangoError):
    """Errors in async job execution."""

Database Operation Exceptions

Specific exceptions for database lifecycle and management operations.

# Database operations
class DatabaseListError(ArangoServerError):
    """Error listing databases."""

class DatabaseCreateError(ArangoServerError):
    """Error creating database."""

class DatabaseDeleteError(ArangoServerError):
    """Error deleting database."""

class DatabasePropertiesError(ArangoServerError):
    """Error getting database properties."""

Collection Operation Exceptions

Exceptions for collection management and document operations.

# Collection management
class CollectionListError(ArangoServerError):
    """Error listing collections."""

class CollectionCreateError(ArangoServerError):
    """Error creating collection."""

class CollectionDeleteError(ArangoServerError):
    """Error deleting collection."""

class CollectionPropertiesError(ArangoServerError):
    """Error getting collection properties."""

class CollectionConfigureError(ArangoServerError):
    """Error configuring collection."""

class CollectionStatisticsError(ArangoServerError):
    """Error getting collection statistics."""

class CollectionRevisionError(ArangoServerError):
    """Error getting collection revision."""

class CollectionChecksumError(ArangoServerError):
    """Error calculating collection checksum."""

# Document operations
class DocumentInsertError(ArangoServerError):
    """Error inserting document."""

class DocumentGetError(ArangoServerError):
    """Error getting document."""

class DocumentUpdateError(ArangoServerError):
    """Error updating document."""

class DocumentReplaceError(ArangoServerError):
    """Error replacing document."""

class DocumentDeleteError(ArangoServerError):
    """Error deleting document."""

class DocumentRevisionError(ArangoServerError):
    """Document revision conflict."""

class DocumentParseError(ArangoClientError):
    """Error parsing document."""

AQL Query Exceptions

Exceptions for AQL query operations, analysis, and management.

# Query execution
class AQLQueryExecuteError(ArangoServerError):
    """Error executing AQL query."""

class AQLQueryExplainError(ArangoServerError):
    """Error explaining AQL query."""

class AQLQueryValidateError(ArangoServerError):
    """Error validating AQL query."""

class AQLQueryListError(ArangoServerError):
    """Error listing running queries."""

class AQLQueryKillError(ArangoServerError):
    """Error killing AQL query."""

class AQLQueryCachePropertiesError(ArangoServerError):
    """Error getting query cache properties."""

class AQLQueryCacheConfigureError(ArangoServerError):
    """Error configuring query cache."""

class AQLQueryCacheClearError(ArangoServerError):
    """Error clearing query cache."""

# User-defined functions
class AQLFunctionListError(ArangoServerError):
    """Error listing AQL functions."""

class AQLFunctionCreateError(ArangoServerError):
    """Error creating AQL function."""

class AQLFunctionDeleteError(ArangoServerError):
    """Error deleting AQL function."""

Graph Operation Exceptions

Exceptions for graph management, vertex, and edge operations.

# Graph management
class GraphListError(ArangoServerError):
    """Error listing graphs."""

class GraphCreateError(ArangoServerError):
    """Error creating graph."""

class GraphDeleteError(ArangoServerError):
    """Error deleting graph."""

class GraphPropertiesError(ArangoServerError):
    """Error getting graph properties."""

# Vertex operations
class VertexCollectionListError(ArangoServerError):
    """Error listing vertex collections."""

class VertexCollectionCreateError(ArangoServerError):
    """Error creating vertex collection."""

class VertexCollectionDeleteError(ArangoServerError):
    """Error deleting vertex collection."""

class VertexGetError(ArangoServerError):
    """Error getting vertex."""

class VertexInsertError(ArangoServerError):
    """Error inserting vertex."""

class VertexUpdateError(ArangoServerError):
    """Error updating vertex."""

class VertexReplaceError(ArangoServerError):
    """Error replacing vertex."""

class VertexDeleteError(ArangoServerError):
    """Error deleting vertex."""

# Edge operations
class EdgeDefinitionListError(ArangoServerError):
    """Error listing edge definitions."""

class EdgeDefinitionCreateError(ArangoServerError):
    """Error creating edge definition."""

class EdgeDefinitionReplaceError(ArangoServerError):
    """Error replacing edge definition."""

class EdgeDefinitionDeleteError(ArangoServerError):
    """Error deleting edge definition."""

class EdgeGetError(ArangoServerError):
    """Error getting edge."""

class EdgeInsertError(ArangoServerError):
    """Error inserting edge."""

class EdgeUpdateError(ArangoServerError):
    """Error updating edge."""

class EdgeReplaceError(ArangoServerError):
    """Error replacing edge."""

class EdgeDeleteError(ArangoServerError):
    """Error deleting edge."""

class EdgeListError(ArangoServerError):
    """Error listing edges."""

class GraphTraverseError(ArangoServerError):
    """Error in graph traversal."""

Index Operation Exceptions

Exceptions for index management and operations.

class IndexListError(ArangoServerError):
    """Error listing indexes."""

class IndexCreateError(ArangoServerError):
    """Error creating index."""

class IndexDeleteError(ArangoServerError):
    """Error deleting index."""

Transaction Exceptions

Exceptions for transaction operations and management.

class TransactionInitError(ArangoServerError):
    """Error initializing transaction."""

class TransactionStatusError(ArangoServerError):
    """Error getting transaction status."""

class TransactionExecuteError(ArangoServerError):
    """Error executing transaction."""

class TransactionCommitError(ArangoServerError):
    """Error committing transaction."""

class TransactionAbortError(ArangoServerError):
    """Error aborting transaction."""

User Management Exceptions

Exceptions for user account operations.

class UserListError(ArangoServerError):
    """Error listing users."""

class UserGetError(ArangoServerError):
    """Error getting user."""

class UserCreateError(ArangoServerError):
    """Error creating user."""

class UserUpdateError(ArangoServerError):
    """Error updating user."""

class UserReplaceError(ArangoServerError):
    """Error replacing user."""

class UserDeleteError(ArangoServerError):
    """Error deleting user."""

class PermissionListError(ArangoServerError):
    """Error listing permissions."""

class PermissionGetError(ArangoServerError):
    """Error getting permission."""

class PermissionUpdateError(ArangoServerError):
    """Error updating permission."""

class PermissionResetError(ArangoServerError):
    """Error resetting permission."""

Error Code Constants

Comprehensive error code constants from ArangoDB server.

# General errors
ERROR_NO_ERROR = 0
ERROR_FAILED = 1
ERROR_SYS_ERROR = 2
ERROR_OUT_OF_MEMORY = 3
ERROR_INTERNAL = 4
ERROR_ILLEGAL_NUMBER = 5
ERROR_NUMERIC_OVERFLOW = 6
ERROR_ILLEGAL_OPTION = 7
ERROR_DEAD_PID = 8
ERROR_NOT_IMPLEMENTED = 9
ERROR_BAD_PARAMETER = 10
ERROR_FORBIDDEN = 11
ERROR_OUT_OF_MEMORY_MMAP = 12
ERROR_CORRUPTED_CSV = 13
ERROR_FILE_NOT_FOUND = 14
ERROR_CANNOT_WRITE_FILE = 15
ERROR_CANNOT_OVERWRITE_FILE = 16
ERROR_TYPE_ERROR = 17
ERROR_LOCK_TIMEOUT = 18
ERROR_CANNOT_CREATE_DIRECTORY = 19
ERROR_CANNOT_CREATE_TEMP_FILE = 20
ERROR_REQUEST_CANCELED = 21
ERROR_DEBUG = 22

# HTTP errors
ERROR_HTTP_BAD_PARAMETER = 400
ERROR_HTTP_UNAUTHORIZED = 401
ERROR_HTTP_FORBIDDEN = 403
ERROR_HTTP_NOT_FOUND = 404
ERROR_HTTP_METHOD_NOT_ALLOWED = 405
ERROR_HTTP_NOT_ACCEPTABLE = 406
ERROR_HTTP_REQUEST_TIMEOUT = 408
ERROR_HTTP_CONFLICT = 409
ERROR_HTTP_GONE = 410
ERROR_HTTP_PRECONDITION_FAILED = 412
ERROR_HTTP_REQUEST_ENTITY_TOO_LARGE = 413
ERROR_HTTP_REQUEST_URI_TOO_LONG = 414
ERROR_HTTP_UNPROCESSABLE_ENTITY = 422
ERROR_HTTP_TOO_MANY_REQUESTS = 429
ERROR_HTTP_REQUEST_HEADER_FIELDS_TOO_LARGE = 431
ERROR_HTTP_INTERNAL_SERVER_ERROR = 500
ERROR_HTTP_NOT_IMPLEMENTED = 501
ERROR_HTTP_BAD_GATEWAY = 502
ERROR_HTTP_SERVICE_UNAVAILABLE = 503
ERROR_HTTP_GATEWAY_TIMEOUT = 504

# ArangoDB specific errors
ERROR_ARANGO_ILLEGAL_STATE = 1000
ERROR_ARANGO_DATAFILE_SEALED = 1002
ERROR_ARANGO_READ_ONLY = 1004
ERROR_ARANGO_DUPLICATE_IDENTIFIER = 1005
ERROR_ARANGO_DATAFILE_UNREADABLE = 1006
ERROR_ARANGO_DATAFILE_EMPTY = 1007
ERROR_ARANGO_RECOVERY = 1008
ERROR_ARANGO_DATAFILE_STATISTICS_NOT_FOUND = 1009

# Document errors
ERROR_ARANGO_CONFLICT = 1200
ERROR_ARANGO_DATADIR_NOT_WRITABLE = 1201
ERROR_ARANGO_DOCUMENT_NOT_FOUND = 1202
ERROR_ARANGO_DATA_SOURCE_NOT_FOUND = 1203
ERROR_ARANGO_COLLECTION_PARAMETER_MISSING = 1204
ERROR_ARANGO_DOCUMENT_HANDLE_BAD = 1205
ERROR_ARANGO_MAXIMAL_SIZE_TOO_SMALL = 1206
ERROR_ARANGO_DUPLICATE_NAME = 1207
ERROR_ARANGO_ILLEGAL_NAME = 1208
ERROR_ARANGO_NO_SUITABLE_INDEX = 1209
ERROR_ARANGO_CROSS_COLLECTION_REQUEST = 1210
ERROR_ARANGO_INDEX_NOT_FOUND = 1212
ERROR_ARANGO_CROSS_COLLECTION_TRANSACTION = 1213
ERROR_ARANGO_INDEX_HANDLE_BAD = 1214
ERROR_ARANGO_DOCUMENT_TOO_LARGE = 1216
ERROR_ARANGO_COLLECTION_NOT_UNLOADED = 1217
ERROR_ARANGO_COLLECTION_TYPE_INVALID = 1218
ERROR_ARANGO_VALIDATION_FAILED = 1219
ERROR_ARANGO_ATTRIBUTE_PARSER_FAILED = 1220
ERROR_ARANGO_DOCUMENT_KEY_BAD = 1221
ERROR_ARANGO_DOCUMENT_KEY_UNEXPECTED = 1222
ERROR_ARANGO_DATADIR_INVALID = 1224
ERROR_ARANGO_DOCUMENT_KEY_MISSING = 1225
ERROR_ARANGO_DOCUMENT_TYPE_INVALID = 1226
ERROR_ARANGO_DATABASE_NOT_FOUND = 1228
ERROR_ARANGO_DATABASE_NAME_INVALID = 1229
ERROR_ARANGO_USE_SYSTEM_DATABASE = 1230
ERROR_ARANGO_ENDPOINT_NOT_FOUND = 1231
ERROR_ARANGO_INVALID_KEY_GENERATOR = 1232
ERROR_ARANGO_INVALID_EDGE_ATTRIBUTE = 1233
ERROR_ARANGO_INDEX_DOCUMENT_ATTRIBUTE_MISSING = 1234
ERROR_ARANGO_INDEX_CREATION_FAILED = 1235
ERROR_ARANGO_WRITE_THROTTLE_TIMEOUT = 1236

# Query errors
ERROR_QUERY_KILLED = 1500
ERROR_QUERY_PARSE = 1501
ERROR_QUERY_EMPTY = 1502
ERROR_QUERY_SCRIPT = 1503
ERROR_QUERY_NUMBER_OUT_OF_RANGE = 1504
ERROR_QUERY_VARIABLE_NAME_INVALID = 1510
ERROR_QUERY_VARIABLE_REDECLARED = 1511
ERROR_QUERY_VARIABLE_NAME_UNKNOWN = 1512
ERROR_QUERY_COLLECTION_LOCK_FAILED = 1521
ERROR_QUERY_TOO_MANY_COLLECTIONS = 1522
ERROR_QUERY_DOCUMENT_ATTRIBUTE_REDECLARED = 1530
ERROR_QUERY_FUNCTION_NAME_UNKNOWN = 1540
ERROR_QUERY_FUNCTION_ARGUMENT_NUMBER_MISMATCH = 1541
ERROR_QUERY_FUNCTION_ARGUMENT_TYPE_MISMATCH = 1542
ERROR_QUERY_INVALID_REGEX = 1543
ERROR_QUERY_BIND_PARAMETERS_INVALID = 1550
ERROR_QUERY_BIND_PARAMETER_MISSING = 1551
ERROR_QUERY_BIND_PARAMETER_UNDECLARED = 1552
ERROR_QUERY_BIND_PARAMETER_TYPE = 1553
ERROR_QUERY_INVALID_LOGICAL_VALUE = 1560
ERROR_QUERY_INVALID_ARITHMETIC_VALUE = 1561
ERROR_QUERY_DIVISION_BY_ZERO = 1562
ERROR_QUERY_ARRAY_EXPECTED = 1563
ERROR_QUERY_FAIL_CALLED = 1569
ERROR_QUERY_GEO_INDEX_MISSING = 1570
ERROR_QUERY_FULLTEXT_INDEX_MISSING = 1571
ERROR_QUERY_INVALID_DATE_VALUE = 1572
ERROR_QUERY_MULTI_MODIFY = 1573
ERROR_QUERY_INVALID_AGGREGATE_EXPRESSION = 1574
ERROR_QUERY_COMPILE_TIME_OPTIONS = 1575
ERROR_QUERY_EXCEPTION_OPTIONS = 1576
ERROR_QUERY_COLLECTION_USED_IN_EXPRESSION = 1577
ERROR_QUERY_DISALLOWED_DYNAMIC_CALL = 1578
ERROR_QUERY_ACCESS_AFTER_MODIFICATION = 1579
ERROR_QUERY_FUNCTION_INVALID_NAME = 1580
ERROR_QUERY_FUNCTION_INVALID_CODE = 1581
ERROR_QUERY_FUNCTION_NOT_FOUND = 1582
ERROR_QUERY_FUNCTION_RUNTIME_ERROR = 1583

# Transaction errors
ERROR_TRANSACTION_INTERNAL = 1650
ERROR_TRANSACTION_NESTED = 1651
ERROR_TRANSACTION_UNREGISTERED_COLLECTION = 1652
ERROR_TRANSACTION_DISALLOWED_OPERATION = 1653
ERROR_TRANSACTION_ABORTED = 1654
ERROR_TRANSACTION_NOT_FOUND = 1655

Type Definitions

Common type aliases and interfaces used throughout the API.

# Type aliases
Json = Dict[str, Any]
"""Dictionary representing JSON object."""

Jsons = List[Json]
"""List of JSON objects."""

Result = Union[T, ArangoError]
"""Result wrapper that contains either success value or error."""

Headers = Optional[MutableMapping[str, str]]
"""HTTP headers mapping."""

DataTypes = Union[str, int, float, bool, None, Dict, List]
"""Supported data types for bind variables."""

Number = Union[int, float]
"""Numeric types."""

# Request/Response types
class Request:
    """HTTP request wrapper."""
    
    @property
    def method(self) -> str:
        """HTTP method."""
    
    @property
    def endpoint(self) -> str:
        """Request endpoint."""
    
    @property
    def headers(self) -> Headers:
        """Request headers."""
    
    @property
    def params(self) -> Optional[Dict[str, str]]:
        """Query parameters."""
    
    @property
    def data(self) -> Any:
        """Request body data."""

class Response:
    """HTTP response wrapper."""
    
    @property
    def method(self) -> str:
        """HTTP method."""
    
    @property
    def url(self) -> str:
        """Request URL."""
    
    @property
    def headers(self) -> Headers:
        """Response headers."""
    
    @property
    def status_code(self) -> int:
        """HTTP status code."""
    
    @property
    def status_text(self) -> str:
        """HTTP status text."""
    
    @property
    def raw_body(self) -> str:
        """Raw response body."""
    
    @property
    def body(self) -> Any:
        """Parsed response body."""
    
    @property
    def error_code(self) -> Optional[int]:
        """ArangoDB error code."""
    
    @property
    def error_message(self) -> Optional[str]:
        """Error message."""
    
    @property
    def is_success(self) -> bool:
        """True if request was successful."""

class Result:
    """Result wrapper for API operations."""
    
    def __init__(self, value: Any = None, error: Optional[ArangoError] = None):
        """Initialize result."""
        
    @property
    def value(self) -> Any:
        """Result value."""
        
    @property
    def error(self) -> Optional[ArangoError]:
        """Error if operation failed."""
        
    @property
    def is_success(self) -> bool:
        """True if operation succeeded."""

Usage Examples

Exception Handling

from arango import (
    ArangoClient,
    ArangoServerError,
    DocumentInsertError,
    CollectionCreateError,
    AQLQueryExecuteError
)

client = ArangoClient()
db = client.db('example', username='root', password='password')

# Handle specific exceptions
try:
    # Try to create collection
    collection = db.create_collection('users')
    
    # Try to insert document
    result = collection.insert({'name': 'Alice', 'age': 25})
    
    # Try to execute query
    cursor = db.aql.execute('FOR u IN users RETURN u')
    users = list(cursor)
    
except CollectionCreateError as e:
    print(f"Failed to create collection: {e.error_message}")
    print(f"Error code: {e.error_code}")
    
except DocumentInsertError as e:
    print(f"Failed to insert document: {e.error_message}")
    if e.error_code == 1210:  # ERROR_ARANGO_CROSS_COLLECTION_REQUEST
        print("Cross-collection request not allowed")
        
except AQLQueryExecuteError as e:
    print(f"Query failed: {e.error_message}")
    print(f"HTTP code: {e.http_code}")
    
except ArangoServerError as e:
    print(f"Server error: {e.error_message}")
    
except Exception as e:
    print(f"Unexpected error: {e}")

Error Code Handling

import arango.errno as errno
from arango import ArangoServerError

def handle_operation_with_error_codes(collection, document):
    try:
        result = collection.insert(document)
        return result
        
    except ArangoServerError as e:
        if e.error_code == errno.ERROR_ARANGO_DUPLICATE_NAME:
            print("Document key already exists")
            # Try update instead
            return collection.update(document)
            
        elif e.error_code == errno.ERROR_ARANGO_DOCUMENT_TOO_LARGE:
            print("Document too large, splitting...")
            # Handle large document
            
        elif e.error_code == errno.ERROR_ARANGO_VALIDATION_FAILED:
            print("Document validation failed")
            print(f"Details: {e.error_message}")
            
        elif e.error_code == errno.ERROR_ARANGO_READ_ONLY:
            print("Database is read-only")
            
        else:
            print(f"Unhandled server error: {e.error_code} - {e.error_message}")
            raise

Custom Error Handling

class DatabaseOperationError(Exception):
    """Custom application-level error."""
    
    def __init__(self, operation: str, arango_error: ArangoError):
        self.operation = operation
        self.arango_error = arango_error
        super().__init__(f"Failed to {operation}: {arango_error.error_message}")

def safe_database_operation(db, operation_func, *args, **kwargs):
    """Wrapper for safe database operations."""
    try:
        return operation_func(*args, **kwargs)
        
    except DocumentInsertError as e:
        raise DatabaseOperationError("insert document", e)
        
    except DocumentUpdateError as e:
        raise DatabaseOperationError("update document", e)
        
    except CollectionCreateError as e:
        raise DatabaseOperationError("create collection", e)
        
    except AQLQueryExecuteError as e:
        raise DatabaseOperationError("execute query", e)
        
    except ArangoServerError as e:
        raise DatabaseOperationError("perform database operation", e)

# Usage
try:
    collection = db.collection('users')
    result = safe_database_operation(
        db, 
        collection.insert, 
        {'name': 'Bob', 'age': 30}
    )
    
except DatabaseOperationError as e:
    print(f"Application error: {e}")
    print(f"Original ArangoDB error: {e.arango_error.error_code}")
    
    # Log for debugging
    import logging
    logging.error(f"Database operation failed", extra={
        'operation': e.operation,
        'error_code': e.arango_error.error_code,
        'error_message': e.arango_error.error_message,
        'http_code': e.arango_error.http_code
    })

Result Pattern Usage

def process_documents_safely(collection, documents):
    """Process documents with comprehensive error handling."""
    results = {
        'success': [],
        'errors': [],
        'skipped': []
    }
    
    for doc in documents:
        try:
            # Validate document first
            if not doc.get('name'):
                results['skipped'].append({
                    'document': doc,
                    'reason': 'Missing required field: name'
                })
                continue
                
            # Try to insert
            result = collection.insert(doc, silent=False)
            results['success'].append({
                'document': doc,
                'key': result['_key'],
                'rev': result['_rev']
            })
            
        except DocumentInsertError as e:
            results['errors'].append({
                'document': doc,
                'error': e.error_message,
                'error_code': e.error_code
            })
            
    return results

# Usage
documents = [
    {'name': 'Alice', 'age': 25},
    {'name': 'Bob'},  # Missing age
    {'age': 30},      # Missing name - will be skipped
    {'name': 'Charlie', 'age': 35}
]

results = process_documents_safely(collection, documents)
print(f"Processed: {len(results['success'])} success, "
      f"{len(results['errors'])} errors, "
      f"{len(results['skipped'])} skipped")

Install with Tessl CLI