tessl/pypi-edgedb
EdgeDB Python driver providing both blocking IO and asyncio implementations for connecting to and interacting with EdgeDB databases.
—
Pending
error-handling.mddocs/
Error Handling
Comprehensive exception hierarchy representing all EdgeDB server errors and client-side issues, providing detailed error context and handling strategies.
Capabilities
Base Error Classes
Foundation classes for all EdgeDB errors and messages.
class EdgeDBError(Exception):
"""
Base class for all EdgeDB errors.
Provides common error attributes and methods for accessing
error details, server context, and error classification.
"""
def has_tag(self, tag: ErrorTag) -> bool:
"""
Check if error has specific tag.
Parameters:
- tag: Error tag to check for
Returns:
True if error has the specified tag
"""
def get_code(self) -> int:
"""
Get error code.
Returns:
Numeric error code from EdgeDB server
"""
def get_server_context(self) -> Optional[str]:
"""
Get server context information.
Returns:
Server-side context information if available
"""
@property
def position(self) -> Optional[int]:
"""Position in query where error occurred."""
@property
def line(self) -> Optional[int]:
"""Line number where error occurred."""
@property
def col(self) -> Optional[int]:
"""Column number where error occurred."""
@property
def hint(self) -> Optional[str]:
"""Hint for resolving the error."""
@property
def details(self) -> Optional[str]:
"""Detailed error information."""
class EdgeDBMessage(Warning):
"""
Base class for EdgeDB messages and warnings.
Represents non-error messages from the database server.
"""
def get_severity(self) -> str:
"""Get message severity level."""
def get_severity_name(self) -> str:
"""Get human-readable severity name."""
def get_code(self) -> int:
"""Get message code."""Server Error Classes
Errors originating from the EdgeDB server.
class InternalServerError(EdgeDBError):
"""Internal server error."""
class UnsupportedFeatureError(EdgeDBError):
"""Feature not supported by the server."""
class ProtocolError(EdgeDBError):
"""Protocol-level communication error."""
class BinaryProtocolError(ProtocolError):
"""Binary protocol parsing error."""
class UnsupportedProtocolVersionError(ProtocolError):
"""Unsupported protocol version."""
class TypeSpecNotFoundError(ProtocolError):
"""Type specification not found."""
class UnexpectedMessageError(ProtocolError):
"""Unexpected protocol message."""
class InputDataError(ProtocolError):
"""Invalid input data format."""
class ParameterTypeMismatchError(InputDataError):
"""Query parameter type mismatch."""
class StateMismatchError(ProtocolError):
"""Client-server state mismatch."""
class ResultCardinalityMismatchError(ProtocolError):
"""Query result cardinality mismatch."""Capability Errors
Errors related to EdgeDB capabilities and features.
class CapabilityError(EdgeDBError):
"""Capability-related error."""
class UnsupportedCapabilityError(CapabilityError):
"""Requested capability not supported."""
class DisabledCapabilityError(CapabilityError):
"""Requested capability is disabled."""Query Errors
Errors in query syntax, semantics, and structure.
class QueryError(EdgeDBError):
"""Base class for query-related errors."""
class InvalidSyntaxError(QueryError):
"""Invalid query syntax."""
class EdgeQLSyntaxError(InvalidSyntaxError):
"""EdgeQL syntax error."""
class SchemaSyntaxError(InvalidSyntaxError):
"""Schema definition syntax error."""
class GraphQLSyntaxError(InvalidSyntaxError):
"""GraphQL syntax error."""
class InvalidTypeError(QueryError):
"""Invalid type usage."""
class InvalidTargetError(InvalidTypeError):
"""Invalid target for operation."""
class InvalidLinkTargetError(InvalidTargetError):
"""Invalid link target."""
class InvalidPropertyTargetError(InvalidTargetError):
"""Invalid property target."""
class InvalidReferenceError(QueryError):
"""Invalid reference in query."""
class UnknownModuleError(InvalidReferenceError):
"""Reference to unknown module."""
class UnknownLinkError(InvalidReferenceError):
"""Reference to unknown link."""
class UnknownPropertyError(InvalidReferenceError):
"""Reference to unknown property."""
class UnknownUserError(InvalidReferenceError):
"""Reference to unknown user."""
class UnknownDatabaseError(InvalidReferenceError):
"""Reference to unknown database."""
class UnknownParameterError(InvalidReferenceError):
"""Reference to unknown parameter."""Schema Errors
Errors in schema definition and modification.
class SchemaError(EdgeDBError):
"""Base class for schema-related errors."""
class SchemaDefinitionError(SchemaError):
"""Schema definition error."""
class InvalidDefinitionError(SchemaDefinitionError):
"""Invalid schema definition."""
class InvalidModuleDefinitionError(InvalidDefinitionError):
"""Invalid module definition."""
class InvalidLinkDefinitionError(InvalidDefinitionError):
"""Invalid link definition."""
class InvalidPropertyDefinitionError(InvalidDefinitionError):
"""Invalid property definition."""
class InvalidUserDefinitionError(InvalidDefinitionError):
"""Invalid user definition."""
class InvalidDatabaseDefinitionError(InvalidDefinitionError):
"""Invalid database definition."""
class InvalidOperatorDefinitionError(InvalidDefinitionError):
"""Invalid operator definition."""
class InvalidAliasDefinitionError(InvalidDefinitionError):
"""Invalid alias definition."""
class InvalidFunctionDefinitionError(InvalidDefinitionError):
"""Invalid function definition."""
class InvalidConstraintDefinitionError(InvalidDefinitionError):
"""Invalid constraint definition."""
class InvalidCastDefinitionError(InvalidDefinitionError):
"""Invalid cast definition."""
class DuplicateDefinitionError(SchemaDefinitionError):
"""Duplicate schema definition."""
class DuplicateModuleDefinitionError(DuplicateDefinitionError):
"""Duplicate module definition."""
class DuplicateLinkDefinitionError(DuplicateDefinitionError):
"""Duplicate link definition."""
class DuplicatePropertyDefinitionError(DuplicateDefinitionError):
"""Duplicate property definition."""
class DuplicateUserDefinitionError(DuplicateDefinitionError):
"""Duplicate user definition."""
class DuplicateDatabaseDefinitionError(DuplicateDefinitionError):
"""Duplicate database definition."""
class DuplicateOperatorDefinitionError(DuplicateDefinitionError):
"""Duplicate operator definition."""
class DuplicateViewDefinitionError(DuplicateDefinitionError):
"""Duplicate view definition."""
class DuplicateFunctionDefinitionError(DuplicateDefinitionError):
"""Duplicate function definition."""
class DuplicateConstraintDefinitionError(DuplicateDefinitionError):
"""Duplicate constraint definition."""
class DuplicateCastDefinitionError(DuplicateDefinitionError):
"""Duplicate cast definition."""
class DuplicateMigrationError(DuplicateDefinitionError):
"""Duplicate migration definition."""Execution Errors
Errors during query and command execution.
class ExecutionError(EdgeDBError):
"""Base class for execution errors."""
class InvalidValueError(ExecutionError):
"""Invalid value for operation."""
class DivisionByZeroError(ExecutionError):
"""Division by zero error."""
class NumericOutOfRangeError(ExecutionError):
"""Numeric value out of range."""
class AccessPolicyError(ExecutionError):
"""Access policy violation."""
class QueryAssertionError(ExecutionError):
"""Query assertion failed."""
class IntegrityError(ExecutionError):
"""Database integrity constraint violation."""
class ConstraintViolationError(IntegrityError):
"""General constraint violation."""
class CardinalityViolationError(ConstraintViolationError):
"""Cardinality constraint violation."""
class MissingRequiredError(ConstraintViolationError):
"""Required value missing."""Transaction Errors
Errors related to transaction management.
class TransactionError(EdgeDBError):
"""Base class for transaction errors."""
class TransactionConflictError(TransactionError):
"""Transaction conflict detected."""
class TransactionSerializationError(TransactionError):
"""Transaction serialization failure."""
class TransactionDeadlockError(TransactionError):
"""Transaction deadlock detected."""
class WatchError(TransactionError):
"""Watch operation error."""Timeout Errors
Errors related to timeouts and time limits.
class SessionTimeoutError(EdgeDBError):
"""Session timeout exceeded."""
class IdleSessionTimeoutError(SessionTimeoutError):
"""Idle session timeout exceeded."""
class QueryTimeoutError(EdgeDBError):
"""Query execution timeout."""
class TransactionTimeoutError(EdgeDBError):
"""Transaction timeout exceeded."""
class IdleTransactionTimeoutError(TransactionTimeoutError):
"""Idle transaction timeout exceeded."""Client Errors
Errors originating from the client side.
class ClientError(EdgeDBError):
"""Base class for client-side errors."""
class ClientConnectionError(ClientError):
"""Client connection error."""
class ClientConnectionFailedError(ClientConnectionError):
"""Connection to server failed."""
class ClientConnectionFailedTemporarilyError(ClientConnectionFailedError):
"""Temporary connection failure."""
class ClientConnectionTimeoutError(ClientConnectionError):
"""Connection timeout."""
class ClientConnectionClosedError(ClientConnectionError):
"""Connection closed unexpectedly."""
class InterfaceError(ClientError):
"""Client interface usage error."""
class QueryArgumentError(InterfaceError):
"""Invalid query arguments."""
class MissingArgumentError(QueryArgumentError):
"""Required argument missing."""
class UnknownArgumentError(QueryArgumentError):
"""Unknown argument provided."""
class InvalidArgumentError(QueryArgumentError):
"""Invalid argument value."""
class NoDataError(InterfaceError):
"""No data returned when data expected."""
class InternalClientError(ClientError):
"""Internal client error."""Access and Authentication Errors
Errors related to access control and authentication.
class ConfigurationError(EdgeDBError):
"""Configuration error."""
class AccessError(EdgeDBError):
"""Access control error."""
class AuthenticationError(AccessError):
"""Authentication failed."""
class AvailabilityError(EdgeDBError):
"""Server availability error."""
class BackendUnavailableError(AvailabilityError):
"""Backend server unavailable."""
class ServerOfflineError(AvailabilityError):
"""Server is offline."""
class BackendError(EdgeDBError):
"""Backend server error."""
class UnsupportedBackendFeatureError(BackendError):
"""Backend feature not supported."""Message Classes
Non-error messages from the database.
class LogMessage(EdgeDBMessage):
"""Log message from server."""
class WarningMessage(EdgeDBMessage):
"""Warning message from server."""Usage Examples
Basic Error Handling
import edgedb
client = edgedb.create_client()
try:
user = client.query_required_single(
"SELECT User { name } FILTER .email = $email",
email="nonexistent@example.com"
)
except edgedb.NoDataError:
print("User not found")
except edgedb.QueryArgumentError as e:
print(f"Invalid query arguments: {e}")
except edgedb.EdgeDBError as e:
print(f"Database error: {e}")Specific Error Types
import edgedb
client = edgedb.create_client()
try:
client.execute("""
INSERT User {
name := 'John Doe',
email := 'john@example.com'
}
""")
except edgedb.ConstraintViolationError as e:
print(f"Constraint violation: {e}")
# Handle duplicate email, etc.
except edgedb.InvalidValueError as e:
print(f"Invalid value: {e}")
# Handle invalid data format
except edgedb.SchemaError as e:
print(f"Schema error: {e}")
# Handle schema-related issuesTransaction Error Handling
import edgedb
client = edgedb.create_client()
max_retries = 3
for attempt in range(max_retries):
try:
with client.transaction() as tx:
tx.execute("UPDATE Account SET balance = balance - 100 WHERE id = $1", account_id)
tx.execute("UPDATE Account SET balance = balance + 100 WHERE id = $2", target_id)
break # Success, exit retry loop
except edgedb.TransactionConflictError:
if attempt == max_retries - 1:
raise # Last attempt failed
print(f"Transaction conflict, retrying... (attempt {attempt + 1})")
except edgedb.TransactionSerializationError:
if attempt == max_retries - 1:
raise
print(f"Serialization error, retrying... (attempt {attempt + 1})")Connection Error Handling
import edgedb
import time
def create_client_with_retry():
max_attempts = 5
base_delay = 1.0
for attempt in range(max_attempts):
try:
return edgedb.create_client()
except edgedb.ClientConnectionFailedError as e:
if attempt == max_attempts - 1:
raise # Last attempt
delay = base_delay * (2 ** attempt) # Exponential backoff
print(f"Connection failed, retrying in {delay}s... (attempt {attempt + 1})")
time.sleep(delay)
except edgedb.ClientConnectionTimeoutError:
print("Connection timeout, check network connectivity")
raise
except edgedb.AuthenticationError:
print("Authentication failed, check credentials")
raise
client = create_client_with_retry()Query Error Analysis
import edgedb
client = edgedb.create_client()
try:
result = client.query("SELECT User { name, email } WHERE .invalid_field = 'value'")
except edgedb.InvalidReferenceError as e:
print(f"Invalid reference: {e}")
print(f"Error position: line {e.line}, column {e.col}")
print(f"Hint: {e.hint}")
except edgedb.EdgeQLSyntaxError as e:
print(f"Syntax error: {e}")
print(f"Error at position {e.position}")
if e.details:
print(f"Details: {e.details}")Comprehensive Error Handler
import edgedb
import logging
logger = logging.getLogger(__name__)
def handle_edgedb_error(e: edgedb.EdgeDBError, operation: str) -> None:
"""Comprehensive EdgeDB error handler."""
error_code = e.get_code() if hasattr(e, 'get_code') else None
if isinstance(e, edgedb.NoDataError):
logger.warning(f"{operation}: No data found")
elif isinstance(e, edgedb.ConstraintViolationError):
logger.error(f"{operation}: Constraint violation - {e}")
elif isinstance(e, edgedb.TransactionConflictError):
logger.warning(f"{operation}: Transaction conflict - {e}")
elif isinstance(e, edgedb.QueryArgumentError):
logger.error(f"{operation}: Invalid arguments - {e}")
elif isinstance(e, edgedb.InvalidSyntaxError):
logger.error(f"{operation}: Syntax error at line {e.line}, col {e.col} - {e}")
elif isinstance(e, edgedb.ClientConnectionError):
logger.error(f"{operation}: Connection error - {e}")
elif isinstance(e, edgedb.AuthenticationError):
logger.error(f"{operation}: Authentication failed - {e}")
elif isinstance(e, edgedb.QueryTimeoutError):
logger.error(f"{operation}: Query timeout - {e}")
else:
logger.error(f"{operation}: Unknown EdgeDB error ({error_code}) - {e}")
# Usage
try:
users = client.query("SELECT User { name, email }")
except edgedb.EdgeDBError as e:
handle_edgedb_error(e, "fetch_users")Context Manager for Error Handling
import edgedb
from contextlib import contextmanager
from typing import Generator, Optional, Type, Union
@contextmanager
def edgedb_error_context(
operation: str,
reraise: bool = True,
ignore_errors: Optional[Union[Type[Exception], tuple]] = None
) -> Generator[None, None, None]:
"""Context manager for EdgeDB error handling."""
try:
yield
except Exception as e:
if ignore_errors and isinstance(e, ignore_errors):
return
if isinstance(e, edgedb.EdgeDBError):
print(f"EdgeDB error in {operation}: {e}")
if hasattr(e, 'hint') and e.hint:
print(f"Hint: {e.hint}")
else:
print(f"Unexpected error in {operation}: {e}")
if reraise:
raise
# Usage
client = edgedb.create_client()
with edgedb_error_context("user_creation"):
client.execute("INSERT User { name := 'Alice', email := 'alice@example.com' }")
# Ignore specific errors
with edgedb_error_context("optional_update", ignore_errors=edgedb.NoDataError):
client.execute("UPDATE User FILTER .email = 'nonexistent@example.com' SET { active := false }")Error Recovery Strategies
import edgedb
from typing import Any, Callable, Optional
def with_fallback(
primary_operation: Callable[[], Any],
fallback_operation: Optional[Callable[[], Any]] = None,
fallback_errors: tuple = (edgedb.NoDataError, edgedb.QueryTimeoutError)
) -> Any:
"""Execute operation with fallback on specific errors."""
try:
return primary_operation()
except fallback_errors as e:
print(f"Primary operation failed ({type(e).__name__}), using fallback")
if fallback_operation:
return fallback_operation()
return None
except edgedb.EdgeDBError:
raise # Re-raise other EdgeDB errors
# Usage
client = edgedb.create_client()
def get_user_expensive():
"""Expensive query with complex computation."""
return client.query_single("SELECT compute_user_stats($id)", id=user_id)
def get_user_basic():
"""Simple fallback query."""
return client.query_single("SELECT User { name } FILTER .id = $id", id=user_id)
user_id = "123e4567-e89b-12d3-a456-426614174000"
user = with_fallback(get_user_expensive, get_user_basic)Install with Tessl CLI
npx tessl i tessl/pypi-edgedb