Source code for langchain.graphs.neptune_graph
from typing import Any, Dict, List, Optional, Tuple, Union
[docs]class NeptuneQueryException(Exception):
    """A class to handle queries that fail to execute"""
    def __init__(self, exception: Union[str, Dict]):
        if isinstance(exception, dict):
            self.message = exception["message"] if "message" in exception else "unknown"
            self.details = exception["details"] if "details" in exception else "unknown"
        else:
            self.message = exception
            self.details = "unknown"
    def get_message(self) -> str:
        return self.message
    def get_details(self) -> Any:
        return self.details
[docs]class NeptuneGraph:
    """Neptune wrapper for graph operations.
    Args:
        host: endpoint for the database instance
        port: port number for the database instance, default is 8182
        use_https: whether to use secure connection, default is True
        client: optional boto3 Neptune client
        credentials_profile_name: optional AWS profile name
        region_name: optional AWS region, e.g., us-west-2
        service: optional service name, default is neptunedata
    Example:
        .. code-block:: python
        graph = NeptuneGraph(
            host='<my-cluster>',
            port=8182
        )
    """
[docs]    def __init__(
        self,
        host: str,
        port: int = 8182,
        use_https: bool = True,
        client: Any = None,
        credentials_profile_name: Optional[str] = None,
        region_name: Optional[str] = None,
        service: str = "neptunedata",
    ) -> None: