解決したい課題

開発者のアクティビティを取得する一環で、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"}}}

おわりに

認証しないと使えません。