<p>使用下面的代码片段,您可以调用函数<code>get_card_list</code>传递身份验证令牌(作为字符串)和管道id(作为整数),并检索管道的整个卡列表</p>
<p><code>get_card_list</code>函数将调用函数<code>request_card_list</code>,直到hasNextpage设置为False,在每次调用中更新光标</p>
<pre><code># 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)
</code></pre>