解決したい課題
開発者のアクティビティを取得する一環で、APIを活用して Openなプルリクエスト上に存在している「未解決スレッド（Un Resolved Conversation）」 をどうにかして取得したい。
なんだけど、GitHub REST API の中にはなさげでした。（ApiVersion：2022-11-28）
解決方法
てなわけで GitHub REST API より柔軟な GitHub GraphQL API を使います。
GitHubのGraphQL APIは、GitHub REST APIよりも正確で柔軟なクエリを提供します。
公式ドキュメント眺めてたらそれっぽいのがありました、やったね。
PullRequestReviewThread > isResolved (Boolean!)
Whether this thread has been resolved.
実行クエリ
isResovledだけ取得しても仕方がないので、すべてのOPENなプルリクエスト かつ 未解決スレッドが存在するものを対象に、今回は以下を一緒に取得します。
- PRのタイトル
- PRの作成者
- レビュワーになってる人
- 未解決スレッドの最後にコメントした人
- 未解決スレッドの最後にコメントされた内容
- 未解決スレッドのURL
QUERY = """
query FetchReviewComments($owner: String!, $repo: String!, $first: Int!, $after: String) {
repository(owner: $owner, name: $repo) {
pullRequests(first: $first, states: OPEN, after: $after) {
edges {
node {
title
author {
login
}
reviewRequests(first: 2) {
nodes {
requestedReviewer {
... on User {
login
}
}
}
}
reviewThreads(last: 15) {
edges {
node {
isResolved
comments(last: 15) {
nodes {
author {
login
}
body
url
}
}
}
}
}
}
}
pageInfo {
endCursor
hasNextPage
}
}
}
}
"""
コード
実装は python（Python 3.12.3）で。CSVに出力するとこまでやります。
以下のコードはほとんど再利用性考えてないので手厳しいレビューは勘弁してください。
GraphQLでUnResolved Conversationを取得する.py
import requests
import csv
# 良い感じに設定
OWNER = "OWNER_NAME"
REPO = "REPOSITORY_NAME"
GITHUB_TOKEN = "PRIVATE_TOKEN"
# ヘッダー
headers = {
"Authorization": f"Bearer {GITHUB_TOKEN}",
"Content-Type": "application/json"
}
# 上記のGraphQLクエリをここに差し込む
QUERY = """
"""
def run_query(query, variables):
response = requests.post('https://api.github.com/graphql',
headers=headers,
json={"query": query, "variables": variables})
if response.ok:
return response.json()
else:
raise Exception(f"Query failed to run by returning code of {response.status_code}, {response.text}")
def write_to_csv(pull_requests):
with open('未解決スレッド.csv', mode='w', newline='', encoding='utf-8') as file:
fieldnames = ['PR_TITLE', 'AUTHOR', 'REVIEWER', 'LAST_COMMENTER', 'UNRESOLVED_THREAD_CONTENT', 'UNRESOLVED_THREAD_URL']
writer = csv.DictWriter(file, fieldnames=fieldnames)
writer.writeheader()
for pull_request in pull_requests:
if 'errors' in pull_request:
print(f"エラーが返されました: {pull_request['errors']}")
continue
else:
# CSVに吐き出すデータ（候補）の初期化
title = pull_request.get("title", "")
author = pull_request.get("author", {}).get("login", "")
reviewer_logins = [
node.get("requestedReviewer", {}).get("login", "")
for node in pull_request.get("reviewRequests", {}).get("nodes", [])
if node.get("requestedReviewer", {}).get("login")
]
extracted_data = {
"PR_TITLE": title,
"AUTHOR": author,
"REVIEWER": ", ".join(reviewer_logins),
"LAST_COMMENTER": "",
"UNRESOLVED_THREAD_CONTENT": "",
"UNRESOLVED_THREAD_URL": ""
}
# 最後の未解決コメントを探す
last_unresolved_comment = None
# レビュースレッドを逆順で調べる
for edge in reversed(pull_request.get("reviewThreads", {}).get("edges", [])):
node = edge.get("node", {})
if not node.get("isResolved"): # 条件：isResolvedキーがfalseである
comments_nodes = node.get("comments", {}).get("nodes", [])
if comments_nodes: # リストの空チェック
last_unresolved_comment = comments_nodes[-1] # 最後のコメントデータを取得
break
# 最後の未解決コメントが見つかった場合、データを抽出
if last_unresolved_comment:
extracted_data["LAST_COMMENTER"] = last_unresolved_comment.get("author", {}).get("login", "")
extracted_data["UNRESOLVED_THREAD_CONTENT"] = last_unresolved_comment.get("body", "")
extracted_data["UNRESOLVED_THREAD_URL"] = last_unresolved_comment.get("url", "")
# CSVに書き込み
writer.writerow(extracted_data)
def fetch_all_pull_request_with_unresolved_thread():
all_pull_requests = []
end_cursor = None # ページネーションのカーソル
while True:
variables = {
"owner": OWNER,
"repo": REPO,
"first": 100, # ここで最初のクエリパラメータをセット
"after": end_cursor # end_cursorがNoneならば最初のページ、そうでなければ次のページを取得
}
result = run_query(QUERY, variables)
repository = result['data']['repository']
pull_requests_data = repository['pullRequests']['edges']
all_pull_requests.extend([pr['node'] for pr in pull_requests_data])
# ページネーション管理
page_info = repository['pullRequests']['pageInfo']
end_cursor = page_info['endCursor']
has_next_page = page_info['hasNextPage']
if not has_next_page:
break
return all_pull_requests
def main():
all_pull_requests = fetch_all_pull_request_with_unresolved_thread()
write_to_csv(all_pull_requests)
if __name__ == "__main__":
main()
実行結果
とれました。
※ 大人の事情で結果は 1 行しか表示してませんが、リポジトリに対するプルリクエストの未解決スレッドすべて取得できるはずです
tips：GraphQL API のレート制限を確認する
Rest APIでもそうですが、API叩きまくってると制限に引っ掛かります。
コードでレート制限を確認する
以下コードの PRIVATE_TOKEN 入れるだけで実行できるはず
あと何回GraphQL叩けるか確認する.py
import requests
GITHUB_TOKEN = "PRIVATE_TOKEN"
# GraphQLクエリとエンドポイント
query = """
{
viewer {
login
}
}
"""
# エンドポイントとヘッダーの準備
endpoint = 'https://api.github.com/graphql'
headers = {
"Authorization": f"Bearer {GITHUB_TOKEN}",
'Content-Type': 'application/json'
}
# クエリ実行
response = requests.post(endpoint, json={'query': query}, headers=headers)
# レートリミット情報の表示
print(f"Rate Limit: {response.headers['X-RateLimit-Limit']}")
print(f"Rate Limit Remaining: {response.headers['X-RateLimit-Remaining']}")
print(f"Rate Limit Reset: {response.headers['X-RateLimit-Reset']} UNIX Time")
print(f"Rate Limit Used: {response.headers['x-RateLimit-Used']}")
# 応答のJSONを表示
print(response.json())
実行結果
TERMINAL
Rate Limit: XXXX
Rate Limit Remaining: XXXX
Rate Limit Reset: XXXXXXXXXX UNIX Time
Rate Limit Used: XX
{'data': {'viewer': {'login': '_mi'}}}
HTTPヘッダーでレート制限を確認
HTTPヘッダを取得する
$ curl -i -X POST -H "Authorization: Bearer <PRIVATE-TOKEN>" -H "Content-Type: application/json" -d '{"query":"{viewer {login}}"}' https://api.github.com/graphql
実行結果
TERMINAL
...
x-ratelimit-limit: XXXX
x-ratelimit-remaining: XXXX
x-ratelimit-reset: XXXXXXXXXX
x-ratelimit-used: XXXX
...
{"data":{"viewer":{"login":"_mi"}}}
おわりに
認証しないと使えません。