Bases: BaseReader
GraphQL 读取器。
将所有 GraphQL 结果合并成 LlamaIndex 使用的文档。
Parameters:
Name |
Type |
Description |
Default |
uri |
str
|
|
None
|
headers |
Optional[Dict]
|
|
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:
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
|