From b7376ad27c51a932e0bcea8934a6edb248a88a81 Mon Sep 17 00:00:00 2001 From: Andy Kwok Date: Wed, 30 Jul 2025 14:36:54 -0700 Subject: [PATCH] Update exceptions Signed-off-by: Andy Kwok --- .../databases/graph/neptune_driver/adapter.py | 6 +- .../graph/neptune_driver/exceptions.py | 78 ++++++++++++++++--- 2 files changed, 71 insertions(+), 13 deletions(-) diff --git a/cognee/infrastructure/databases/graph/neptune_driver/adapter.py b/cognee/infrastructure/databases/graph/neptune_driver/adapter.py index 664e2b9f5..eb9f65004 100644 --- a/cognee/infrastructure/databases/graph/neptune_driver/adapter.py +++ b/cognee/infrastructure/databases/graph/neptune_driver/adapter.py @@ -71,10 +71,10 @@ class NeptuneGraphDB(GraphDBInterface): # Validate configuration if not validate_graph_id(graph_id): - raise NeptuneAnalyticsConfigurationError(f"Invalid graph ID: \"{graph_id}\"") + raise NeptuneAnalyticsConfigurationError(message=f"Invalid graph ID: \"{graph_id}\"") if region and not validate_aws_region(region): - raise NeptuneAnalyticsConfigurationError(f"Invalid AWS region: \"{region}\"") + raise NeptuneAnalyticsConfigurationError(message=f"Invalid AWS region: \"{region}\"") self.graph_id = graph_id self.region = region @@ -123,7 +123,7 @@ class NeptuneGraphDB(GraphDBInterface): return client except Exception as e: - raise NeptuneAnalyticsConfigurationError(f"Failed to initialize Neptune Analytics client: {format_neptune_error(e)}") from e + raise NeptuneAnalyticsConfigurationError(message=f"Failed to initialize Neptune Analytics client: {format_neptune_error(e)}") from e @staticmethod def _serialize_properties(properties: Dict[str, Any]) -> Dict[str, Any]: diff --git a/cognee/infrastructure/databases/graph/neptune_driver/exceptions.py b/cognee/infrastructure/databases/graph/neptune_driver/exceptions.py index 984f2c9f0..10aa0bb4b 100644 --- a/cognee/infrastructure/databases/graph/neptune_driver/exceptions.py +++ b/cognee/infrastructure/databases/graph/neptune_driver/exceptions.py @@ -2,48 +2,106 @@ This module defines custom exceptions for Neptune Analytics operations. """ +from cognee.exceptions import CogneeApiError +from fastapi import status -class NeptuneAnalyticsError(Exception): +class NeptuneAnalyticsError(CogneeApiError): """Base exception for Neptune Analytics operations.""" - pass + def __init__( + self, + message: str = "Neptune Analytics error.", + name: str = "NeptuneAnalyticsError", + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR + ): + super().__init__(message, name, status_code) + class NeptuneAnalyticsConnectionError(NeptuneAnalyticsError): """Exception raised when connection to Neptune Analytics fails.""" - pass + def __init__( + self, + message: str = "Unable to connect to Neptune Analytics. Please check the endpoint and network connectivity.", + name: str = "NeptuneAnalyticsConnectionError", + status_code=status.HTTP_404_NOT_FOUND + ): + super().__init__(message, name, status_code) class NeptuneAnalyticsQueryError(NeptuneAnalyticsError): """Exception raised when a query execution fails.""" - pass + def __init__( + self, + message: str = "The query execution failed due to invalid syntax or semantic issues.", + name: str = "NeptuneAnalyticsQueryError", + status_code=status.HTTP_400_BAD_REQUEST + ): + super().__init__(message, name, status_code) class NeptuneAnalyticsAuthenticationError(NeptuneAnalyticsError): """Exception raised when authentication with Neptune Analytics fails.""" - pass + def __init__( + self, + message: str = "Authentication with Neptune Analytics failed. Please verify your credentials.", + name: str = "NeptuneAnalyticsAuthenticationError", + status_code=status.HTTP_400_BAD_REQUEST + ): + super().__init__(message, name, status_code) class NeptuneAnalyticsConfigurationError(NeptuneAnalyticsError): """Exception raised when Neptune Analytics configuration is invalid.""" - pass + def __init__( + self, + message: str = "Neptune Analytics configuration is invalid or incomplete. Please review your setup.", + name: str = "NeptuneAnalyticsConfigurationError", + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR + ): + super().__init__(message, name, status_code) class NeptuneAnalyticsTimeoutError(NeptuneAnalyticsError): """Exception raised when a Neptune Analytics operation times out.""" - pass + def __init__( + self, + message: str = "The operation timed out while communicating with Neptune Analytics.", + name: str = "NeptuneAnalyticsTimeoutError", + status_code=status.HTTP_504_GATEWAY_TIMEOUT + ): + super().__init__(message, name, status_code) class NeptuneAnalyticsThrottlingError(NeptuneAnalyticsError): """Exception raised when requests are throttled by Neptune Analytics.""" - pass + def __init__( + self, + message: str = "Request was throttled by Neptune Analytics due to exceeding rate limits.", + name: str = "NeptuneAnalyticsThrottlingError", + status_code=status.HTTP_429_TOO_MANY_REQUESTS + ): + super().__init__(message, name, status_code) class NeptuneAnalyticsResourceNotFoundError(NeptuneAnalyticsError): """Exception raised when a Neptune Analytics resource is not found.""" - pass + def __init__( + self, + message: str = "The requested Neptune Analytics resource could not be found.", + name: str = "NeptuneAnalyticsResourceNotFoundError", + status_code=status.HTTP_404_NOT_FOUND + ): + super().__init__(message, name, status_code) class NeptuneAnalyticsInvalidParameterError(NeptuneAnalyticsError): """Exception raised when invalid parameters are provided to Neptune Analytics.""" - pass + def __init__( + self, + message: str = "One or more parameters provided to Neptune Analytics are invalid or missing.", + name: str = "NeptuneAnalyticsInvalidParameterError", + status_code=status.HTTP_400_BAD_REQUEST + ): + super().__init__(message, name, status_code) +