Skip to content

Graphql

GraphQLReader #

Bases: BaseReader

GraphQL 读取器。

将所有 GraphQL 结果合并成 LlamaIndex 使用的文档。

Parameters:

Name Type Description Default
uri str

GraphQL 地址。

None
headers Optional[Dict]

可选的 HTTP 头部。

None
Source code in llama_index/readers/graphql/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class GraphQLReader(BaseReader):
    """GraphQL 读取器。

将所有 GraphQL 结果合并成 LlamaIndex 使用的文档。

Args:
    uri (str): GraphQL 地址。
    headers (Optional[Dict]): 可选的 HTTP 头部。"""

    def __init__(
        self,
        uri: Optional[str] = None,
        headers: Optional[Dict] = None,
    ) -> None:
        """使用参数进行初始化。"""
        try:
            from gql import Client
            from gql.transport.requests import RequestsHTTPTransport

        except ImportError:
            raise ImportError("`gql` package not found, please run `pip install gql`")
        if uri:
            if uri is None:
                raise ValueError("`uri` must be provided.")
            if headers is None:
                headers = {}
            transport = RequestsHTTPTransport(url=uri, headers=headers)
            self.client = Client(transport=transport, fetch_schema_from_transport=True)

    def load_data(self, query: str, variables: Optional[Dict] = None) -> List[Document]:
        """运行带有可选变量的查询,并将结果转换为文档。

Args:
    query(str):GraphQL查询字符串。
    variables(Optional[Dict]):可选的查询参数。

Returns:
    List[Document]:文档列表。
"""
        try:
            from gql import gql

        except ImportError:
            raise ImportError("`gql` package not found, please run `pip install gql`")
        if variables is None:
            variables = {}

        documents = []

        result = self.client.execute(gql(query), variable_values=variables)

        for key in result:
            entry = result[key]
            if isinstance(entry, list):
                documents.extend([Document(text=yaml.dump(v)) for v in entry])
            else:
                documents.append(Document(text=yaml.dump(entry)))

        return documents

load_data #

load_data(
    query: str, variables: Optional[Dict] = None
) -> List[Document]

运行带有可选变量的查询,并将结果转换为文档。

Returns:

Type Description
List[Document]

List[Document]:文档列表。

Source code in llama_index/readers/graphql/base.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
    def load_data(self, query: str, variables: Optional[Dict] = None) -> List[Document]:
        """运行带有可选变量的查询,并将结果转换为文档。

Args:
    query(str):GraphQL查询字符串。
    variables(Optional[Dict]):可选的查询参数。

Returns:
    List[Document]:文档列表。
"""
        try:
            from gql import gql

        except ImportError:
            raise ImportError("`gql` package not found, please run `pip install gql`")
        if variables is None:
            variables = {}

        documents = []

        result = self.client.execute(gql(query), variable_values=variables)

        for key in result:
            entry = result[key]
            if isinstance(entry, list):
                documents.extend([Document(text=yaml.dump(v)) for v in entry])
            else:
                documents.append(Document(text=yaml.dump(entry)))

        return documents