Python GraphQL 查询问题

Question

我正在使用 Python 向 Pipefy GraphQL API 发出请求。 我已经阅读了文档并在 pipefy 论坛中进行了搜索，但是 我不知道下面的查询有什么问题：

pipeId = '171258'
query ="""
        {
            "query": "{allCards(pipeId: %s, first: 30, after: 'WyIxLjAiLCI1ODAuMCIsMzI0OTU0NF0'){pageInfo{endCursor hasNextPage}edges{node{id title}}}}"
        }
        """%(pipeid)

在我添加after 参数之前，查询运行良好。 我已经尝试过类似的变体：

after: "WyIxLjAiLCI1ODAuMCIsMzI0OTU0NF0"

after: \"WyIxLjAiLCI1ODAuMCIsMzI0OTU0NF0\"

after: \n"WyIxLjAiLCI1ODAuMCIsMzI0OTU0NF0\n"

我知道问题与转义有关，因为 API 返回如下消息：

'{"errors":[{"locations":[{"column":45,"line":1}],"message":"token recognition error at: \'\'\'"},{"locations":[{"column":77,"line":1}],"message":"token recognition error at: \'\'\'"}]}\n'

（使用after: 'WyIxLjAiLCI1ODAuMCIsMzI0OTU0NF0'发出请求时返回此消息）

这里的任何帮助都非常少！ 谢谢

Answer 1

我今天遇到了和你一样的问题（在 Pipefy 的支持页面上看到了你的帖子）。我亲自接触过 Pipefy 的开发人员，但他们一点帮助都没有。

我通过正确转义查询来解决它。

试试这样：

query = '{"query": "{ allCards(pipeId: %s, first: 30, after: \\"WyIxLjAiLCI1ODAuMCIsMzI0OTU0NF0\\"){ pageInfo{endCursor hasNextPage } edges { node { id title } } } }"}'

在光标中包含的双引号之前使用单引号和双反斜杠定义字符串。

Answer 2

使用下面的代码 sn-p，您可以调用函数 get_card_list，传递身份验证令牌（作为字符串）和 pipe_id（作为整数）并检索管道的整个卡列表。

get_card_list 函数将调用函数request_card_list 直到 hasNextpage 设置为 False，每次调用时更新光标。

# Function responsible to get cards from a pipe using Pipefy's GraphQL API
def request_card_list(auth_token, pipe_id, hasNextPage=False, endCursor=""):    
    url = "https://api.pipefy.com/graphql"
    headers = {
        'Content-Type': 'application/json',
        'Authorization': 'Bearer %s' %auth_token
    }
    if not hasNextPage:
        payload = '{"query": "{ allCards(pipeId: %i, first: 50) { edges { node { id title  phases_history { phase { name } firstTimeIn lastTimeOut } } cursor } pageInfo { endCursor hasNextPage } } }"}' %pipe_id
    else:
        payload = '{"query": "{ allCards(pipeId: %i, first: 50, after: \\"%s\\") { edges { node { id title  phases_history { phase { name } firstTimeIn lastTimeOut } } cursor } pageInfo { endCursor hasNextPage } } }"}' % (pipe_id, endCursor)

    response = requests.request("POST", url, data=payload, headers=headers)
    response_body = response.text
    response_body_dict = json.loads(response_body)
    response_dict_list = response_body_dict['data']['allCards']['edges']

    card_list = []
    for d in response_dict_list:
        for h in d['node']['phases_history']:
            h['firstTimeIn'] = datetime.strptime(h['firstTimeIn'], date_format)
            if h['lastTimeOut']:
                h['lastTimeOut'] = datetime.strptime(h['lastTimeOut'], date_format)
        card_list.append(d['node'])

    return_list = [card_list, response_body_dict['data']['allCards']['pageInfo']['hasNextPage'], response_body_dict['data']['allCards']['pageInfo']['endCursor']]
    return return_list

# Function responsible to get all cards from a pipe using Pipefy's GraphQL API and pagination
def get_card_list(auth_token, pipe_id):
    card_list = []
    response = request_card_list(auth_token, pipe_id)
    card_list = card_list + response[0]

    while response[1]:
        response = request_card_list(auth_token, pipe_id, response[1], response[2])
        card_list = card_list + response[0]

    return(card_list)

Answer 3

感谢 Lodi 的回答，我可以进行下一步了。 如何使用变量传递查询的“after”参数 由于难度很大，所以决定在这里分享给那些面临同样挑战的人。

end_cursor = 'WyIxLjAiLCI2NTcuMCIsNDgwNDA2OV0'
end_cursor = "\\" + "\"" + end_cursor  + "\\" + "\""
# desired output: end_cursor = '\"WyIxLjAiLCI2NTcuMCIsNDgwNDA2OV0\"'
query ="""
       {
        "query": "{allCards(pipeId: %s, first: 50, after: %s){pageInfo{endCursor hasNextPage}edges{node{id title}}}}"
        }
       """%(pipeid, end_cursor)