SelfMemory

"""Structured exception classes for SelfMemory with error codes, suggestions, and debug information. This module provides a comprehensive set of exception classes that replace the generic APIError with specific, actionable exceptions. Each exception includes error codes, user-friendly suggestions, and debug information to enable better error handling and recovery in applications using SelfMemory. Example: Basic usage: try: memory.add(content, user_id=user_id) except RateLimitError as e: # Implement exponential backoff time.sleep(e.debug_info.get('retry_after', 60)) except MemoryQuotaExceededError as e: # Trigger quota upgrade flow logger.error(f"Quota exceeded: {e.error_code}") except ValidationError as e: # Return user-friendly error raise HTTPException(400, detail=e.suggestion) Advanced usage with error context: try: memory.update(memory_id, content=new_content) except MemoryNotFoundError as e: logger.warning(f"Memory {memory_id} not found: {e.message}") if e.suggestion: logger.info(f"Suggestion: {e.suggestion}") """ from typing import Any class MemoryError(Exception): """Base exception for all memory-related errors. This is the base class for all SelfMemory-specific exceptions. It provides a structured approach to error handling with error codes, contextual details, suggestions for resolution, and debug information. Attributes: message (str): Human-readable error message. error_code (str): Unique error identifier for programmatic handling. details (dict): Additional context about the error. suggestion (str): User-friendly suggestion for resolving the error. debug_info (dict): Technical debugging information. Example: raise MemoryError( message="Memory operation failed", error_code="MEM_001", details={"operation": "add", "user_id": "user123"}, suggestion="Please check your API key and try again", debug_info={"request_id": "req_456", "timestamp": "2024-01-01T00:00:00Z"} ) """ def __init__( self, message: str, error_code: str, details: dict[str, Any] | None = None, suggestion: str | None = None, debug_info: dict[str, Any] | None = None, ): """Initialize a MemoryError. Args: message: Human-readable error message. error_code: Unique error identifier. details: Additional context about the error. suggestion: User-friendly suggestion for resolving the error. debug_info: Technical debugging information. """ self.message = message self.error_code = error_code self.details = details or {} self.suggestion = suggestion self.debug_info = debug_info or {} super().__init__(self.message) def __repr__(self) -> str: return ( f"{self.__class__.__name__}(" f"message={self.message!r}, " f"error_code={self.error_code!r}, " f"details={self.details!r}, " f"suggestion={self.suggestion!r}, " f"debug_info={self.debug_info!r})" ) class AuthenticationError(MemoryError): """Raised when authentication fails. This exception is raised when API key validation fails, tokens are invalid, or authentication credentials are missing or expired. Common scenarios: - Invalid API key - Expired authentication token - Missing authentication headers - Insufficient permissions Example: raise AuthenticationError( message="Invalid API key provided", error_code="AUTH_001", suggestion="Please check your API key in the SelfMemory dashboard" ) """ pass class RateLimitError(MemoryError): """Raised when rate limits are exceeded. This exception is raised when the API rate limit has been exceeded. It includes information about retry timing and current rate limit status. The debug_info typically contains: - retry_after: Seconds to wait before retrying - limit: Current rate limit - remaining: Remaining requests in current window - reset_time: When the rate limit window resets Example: raise RateLimitError( message="Rate limit exceeded", error_code="RATE_001", suggestion="Please wait before making more requests", debug_info={"retry_after": 60, "limit": 100, "remaining": 0} ) """ pass class ValidationError(MemoryError): """Raised when input validation fails. This exception is raised when request parameters, memory content, or configuration values fail validation checks. Common scenarios: - Invalid user_id format - Missing required fields - Content too long or too short - Invalid metadata format - Malformed filters Example: raise ValidationError( message="Invalid user_id format", error_code="VAL_001", details={"field": "user_id", "value": "123", "expected": "string"}, suggestion="User ID must be a non-empty string" ) """ pass class MemoryNotFoundError(MemoryError): """Raised when a memory is not found. This exception is raised when attempting to access, update, or delete a memory that doesn't exist or is not accessible to the current user. Example: raise MemoryNotFoundError( message="Memory not found", error_code="MEM_404", details={"memory_id": "mem_123", "user_id": "user_456"}, suggestion="Please check the memory ID and ensure it exists" ) """ pass class NetworkError(MemoryError): """Raised when network connectivity issues occur. This exception is raised for network-related problems such as connection timeouts, DNS resolution failures, or service unavailability. Common scenarios: - Connection timeout - DNS resolution failure - Service temporarily unavailable - Network connectivity issues Example: raise NetworkError( message="Connection timeout", error_code="NET_001", suggestion="Please check your internet connection and try again", debug_info={"timeout": 30, "endpoint": "api.selfmemory.com"} ) """ pass class ConfigurationError(MemoryError): """Raised when client configuration is invalid. This exception is raised when the client is improperly configured, such as missing required settings or invalid configuration values. Common scenarios: - Missing API key - Invalid host URL - Incompatible configuration options - Missing required environment variables Example: raise ConfigurationError( message="API key not configured", error_code="CFG_001", suggestion="Set SELFMEMORY_API_KEY environment variable or pass api_key parameter" ) """ pass class MemoryQuotaExceededError(MemoryError): """Raised when user's memory quota is exceeded. This exception is raised when the user has reached their memory storage or usage limits. The debug_info typically contains: - current_usage: Current memory usage - quota_limit: Maximum allowed usage - usage_type: Type of quota (storage, requests, etc.) Example: raise MemoryQuotaExceededError( message="Memory quota exceeded", error_code="QUOTA_001", suggestion="Please upgrade your plan or delete unused memories", debug_info={"current_usage": 1000, "quota_limit": 1000, "usage_type": "memories"} ) """ pass class MemoryCorruptionError(MemoryError): """Raised when memory data is corrupted. This exception is raised when stored memory data is found to be corrupted, malformed, or otherwise unreadable. Example: raise MemoryCorruptionError( message="Memory data is corrupted", error_code="CORRUPT_001", details={"memory_id": "mem_123"}, suggestion="Please contact support for data recovery assistance" ) """ pass class VectorSearchError(MemoryError): """Raised when vector search operations fail. This exception is raised when vector database operations fail, such as search queries, embedding generation, or index operations. Common scenarios: - Embedding model unavailable - Vector index corruption - Search query timeout - Incompatible vector dimensions Example: raise VectorSearchError( message="Vector search failed", error_code="VEC_001", details={"query": "find similar memories", "vector_dim": 1536}, suggestion="Please try a simpler search query" ) """ pass class CacheError(MemoryError): """Raised when caching operations fail. This exception is raised when cache-related operations fail, such as cache misses, cache invalidation errors, or cache corruption. Example: raise CacheError( message="Cache operation failed", error_code="CACHE_001", details={"operation": "get", "key": "user_memories_123"}, suggestion="Cache will be refreshed automatically" ) """ pass # OSS-specific exception classes class VectorStoreError(MemoryError): """Raised when vector store operations fail. This exception is raised when vector store operations fail, such as embedding storage, similarity search, or vector operations. Example: raise VectorStoreError( message="Vector store operation failed", error_code="VECTOR_001", details={"operation": "search", "collection": "memories"}, suggestion="Please check your vector store configuration and connection" ) """ def __init__( self, message: str, error_code: str = "VECTOR_001", details: dict = None, suggestion: str = "Please check your vector store configuration and connection", debug_info: dict = None, ): super().__init__(message, error_code, details, suggestion, debug_info) class GraphStoreError(MemoryError): """Raised when graph store operations fail. This exception is raised when graph store operations fail, such as relationship creation, entity management, or graph queries. Example: raise GraphStoreError( message="Graph store operation failed", error_code="GRAPH_001", details={"operation": "create_relationship", "entity": "user_123"}, suggestion="Please check your graph store configuration and connection" ) """ def __init__( self, message: str, error_code: str = "GRAPH_001", details: dict = None, suggestion: str = "Please check your graph store configuration and connection", debug_info: dict = None, ): super().__init__(message, error_code, details, suggestion, debug_info) class EmbeddingError(MemoryError): """Raised when embedding operations fail. This exception is raised when embedding operations fail, such as text embedding generation or embedding model errors. Example: raise EmbeddingError( message="Embedding generation failed", error_code="EMBED_001", details={"text_length": 1000, "model": "openai"}, suggestion="Please check your embedding model configuration" ) """ def __init__( self, message: str, error_code: str = "EMBED_001", details: dict = None, suggestion: str = "Please check your embedding model configuration", debug_info: dict = None, ): super().__init__(message, error_code, details, suggestion, debug_info) class LLMError(MemoryError): """Raised when LLM operations fail. This exception is raised when LLM operations fail, such as text generation, completion, or model inference errors. Example: raise LLMError( message="LLM operation failed", error_code="LLM_001", details={"model": "gpt-4", "prompt_length": 500}, suggestion="Please check your LLM configuration and API key" ) """ def __init__( self, message: str, error_code: str = "LLM_001", details: dict = None, suggestion: str = "Please check your LLM configuration and API key", debug_info: dict = None, ): super().__init__(message, error_code, details, suggestion, debug_info) class DatabaseError(MemoryError): """Raised when database operations fail. This exception is raised when database operations fail, such as SQLite operations, connection issues, or data corruption. Example: raise DatabaseError( message="Database operation failed", error_code="DB_001", details={"operation": "insert", "table": "memories"}, suggestion="Please check your database configuration and connection" ) """ def __init__( self, message: str, error_code: str = "DB_001", details: dict = None, suggestion: str = "Please check your database configuration and connection", debug_info: dict = None, ): super().__init__(message, error_code, details, suggestion, debug_info) class DependencyError(MemoryError): """Raised when required dependencies are missing. This exception is raised when required dependencies are missing, such as optional packages for specific providers or features. Example: raise DependencyError( message="Required dependency missing", error_code="DEPS_001", details={"package": "kuzu", "feature": "graph_store"}, suggestion="Please install the required dependencies: pip install kuzu" ) """ def __init__( self, message: str, error_code: str = "DEPS_001", details: dict = None, suggestion: str = "Please install the required dependencies", debug_info: dict = None, ): super().__init__(message, error_code, details, suggestion, debug_info) # Mapping of HTTP status codes to specific exception classes HTTP_STATUS_TO_EXCEPTION = { 400: ValidationError, 401: AuthenticationError, 403: AuthenticationError, 404: MemoryNotFoundError, 408: NetworkError, 409: ValidationError, 413: MemoryQuotaExceededError, 422: ValidationError, 429: RateLimitError, 500: MemoryError, 502: NetworkError, 503: NetworkError, 504: NetworkError, } def create_exception_from_response( status_code: int, response_text: str, error_code: str | None = None, details: dict[str, Any] | None = None, debug_info: dict[str, Any] | None = None, ) -> MemoryError: """Create an appropriate exception based on HTTP response. This function analyzes the HTTP status code and response to create the most appropriate exception type with relevant error information. Args: status_code: HTTP status code from the response. response_text: Response body text. error_code: Optional specific error code. details: Additional error context. debug_info: Debug information. Returns: An instance of the appropriate MemoryError subclass. Example: exception = create_exception_from_response( status_code=429, response_text="Rate limit exceeded", debug_info={"retry_after": 60} ) # Returns a RateLimitError instance """ exception_class = HTTP_STATUS_TO_EXCEPTION.get(status_code, MemoryError) # Generate error code if not provided if not error_code: error_code = f"HTTP_{status_code}" # Create appropriate suggestion based on status code suggestions = { 400: "Please check your request parameters and try again", 401: "Please check your API key and authentication credentials", 403: "You don't have permission to perform this operation", 404: "The requested resource was not found", 408: "Request timed out. Please try again", 409: "Resource conflict. Please check your request", 413: "Request too large. Please reduce the size of your request", 422: "Invalid request data. Please check your input", 429: "Rate limit exceeded. Please wait before making more requests", 500: "Internal server error. Please try again later", 502: "Service temporarily unavailable. Please try again later", 503: "Service unavailable. Please try again later", 504: "Gateway timeout. Please try again later", } suggestion = suggestions.get(status_code, "Please try again later") return exception_class( message=response_text or f"HTTP {status_code} error", error_code=error_code, details=details or {}, suggestion=suggestion, debug_info=debug_info or {}, )