Source code for langchain.graphs.neptune_graph
import json
from typing import Any, Dict, List, Tuple, Union
import requests
[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. This version
    does not support Sigv4 signing of requests.
    Example:
        .. code-block:: python
        graph = NeptuneGraph(
            host='<my-cluster>',
            port=8182
        )
    """
[docs]    def __init__(self, host: str, port: int = 8182, use_https: bool = True) -> None:
        """Create a new Neptune graph wrapper instance."""
        if use_https:
            self.summary_url = (
                f"https://{host}:{port}/pg/statistics/summary?mode=detailed"
            )
            self.query_url = f"https://{host}:{port}/openCypher"
        else:
            self.summary_url = (
                f"http://{host}:{port}/pg/statistics/summary?mode=detailed"
            )
            self.query_url = f"http://{host}:{port}/openCypher"
        # Set schema
        try:
            self._refresh_schema()
        except NeptuneQueryException: