Source code for langchain_community.graphs.tigergraph_graph

from typing import Any, Dict, List, Optional

from langchain_community.graphs.graph_store import GraphStore


class TigerGraph(GraphStore):
    """TigerGraph图操作的包装器。

    *安全提示*：确保数据库连接使用的凭据范围狭窄，仅包括必要的权限。
        如果未这样做，可能会导致数据损坏或丢失，因为调用代码可能会尝试执行会导致删除、变异数据（如果适当提示）或读取敏感数据（如果数据库中存在此类数据）的命令。
        防范这些负面结果的最佳方法是（根据情况）限制授予此工具使用的凭据的权限。

        有关更多信息，请参见 https://python.langchain.com/docs/security。"""

    def __init__(self, conn: Any) -> None:
        """创建一个新的TigerGraph图包装器实例。"""
        self.set_connection(conn)
        self.set_schema()

    @property
    def conn(self) -> Any:
        return self._conn

    @property
    def schema(self) -> Dict[str, Any]:
        return self._schema

    def get_schema(self) -> str:  # type: ignore[override]
        if self._schema:
            return str(self._schema)
        else:
            self.set_schema()
            return str(self._schema)

    def set_connection(self, conn: Any) -> None:
        try:
            from pyTigerGraph import TigerGraphConnection
        except ImportError:
            raise ImportError(
                "Could not import pyTigerGraph python package. "
                "Please install it with `pip install pyTigerGraph`."
            )

        if not isinstance(conn, TigerGraphConnection):
            msg = "**conn** parameter must inherit from TigerGraphConnection"
            raise TypeError(msg)

        if conn.ai.nlqs_host is None:
            msg = """**conn** parameter does not have nlqs_host parameter defined.
                     Define hostname of NLQS service."""
            raise ConnectionError(msg)

        self._conn: TigerGraphConnection = conn
        self.set_schema()

    def set_schema(self, schema: Optional[Dict[str, Any]] = None) -> None:
        """设置TigerGraph数据库的模式。
如果 **schema** 为None，则自动生成模式。
"""
        self._schema = self.generate_schema() if schema is None else schema

    def generate_schema(
        self,
    ) -> Dict[str, List[Dict[str, Any]]]:
        """生成TigerGraph数据库的模式并返回
用户可以指定一个 **sample_ratio** （0到1）来确定
用于呈现每个Collection模式的文档/边的比例
（相对于Collection大小）。
"""
        return self._conn.getSchema(force=True)

    def refresh_schema(self):  # type: ignore[no-untyped-def]
        self.generate_schema()

    def query(self, query: str) -> Dict[str, Any]:  # type: ignore[override]
        """查询TigerGraph数据库。"""
        answer = self._conn.ai.query(query)
        return answer

    def register_query(
        self,
        function_header: str,
        description: str,
        docstring: str,
        param_types: dict = {},
    ) -> List[str]:
        """
        包装函数，用于将自定义的GSQL查询注册到TigerGraph NLQS。
        """
        return self._conn.ai.registerCustomQuery(
            function_header, description, docstring, param_types
        )