Source code for langchain.graphs.nebula_graph
import logging
from string import Template
from typing import Any, Dict, Optional
logger = logging.getLogger(__name__)
rel_query = Template(
    """
MATCH ()-[e:`$edge_type`]->()
  WITH e limit 1
MATCH (m)-[:`$edge_type`]->(n) WHERE id(m) == src(e) AND id(n) == dst(e)
RETURN "(:" + tags(m)[0] + ")-[:$edge_type]->(:" + tags(n)[0] + ")" AS rels
"""
)
RETRY_TIMES = 3
[docs]class NebulaGraph:
    """NebulaGraph wrapper for graph operations
    NebulaGraph inherits methods from Neo4jGraph to bring ease to the user space.
    """
[docs]    def __init__(
        self,
        space: str,
        username: str = "root",
        password: str = "nebula",
        address: str = "127.0.0.1",
        port: int = 9669,
        session_pool_size: int = 30,
    ) -> None:
        """Create a new NebulaGraph wrapper instance."""
        try:
            import nebula3  # noqa: F401
            import pandas  # noqa: F401
        except ImportError:
            raise ValueError(
                "Please install NebulaGraph Python client and pandas first: "
                "`pip install nebula3-python pandas`"
            )
        self.username = username
        self.password = password
        self.address = address
        self.port = port
        self.space = space
        self.session_pool_size = session_pool_size
        self.session_pool = self._get_session_pool()
        self.schema = ""