newModuleTagger: use graphql API, pagination

newModuleTagger/find_prs.py

@@ -1,10 +1,63 @@
 #! /usr/bin/env nix-shell
-#! nix-shell -i python3 -p python3 gh
+#! nix-shell -i python3 -p python3 gh python3Packages.tqdm
 
+from tqdm import tqdm
 from pathlib import Path
 import subprocess
 import json
 
+
+PAGINATION_STEP=20
+
+
+GRAPHQL_QUERY = """
+query($endCursor: String) {
+  repository(owner: "NixOS", name: "nixpkgs") {
+    pullRequests(
+      first: """ + str(PAGINATION_STEP) + """,
+      after: $endCursor,
+      orderBy: {
+        field: CREATED_AT,
+        direction: DESC
+    	},
+      labels: ["8.has: module (update)"],
+    ) {
+      edges {
+        node {
+          title
+          number
+          state
+          createdAt
+          author {
+            login
+          }
+          files(first: 50) {
+            edges {
+              node {
+                path
+                changeType
+              }
+            }
+          }
+          labels(first: 50) {
+            edges {
+              node {
+                name
+              }
+            }
+          }
+        }
+      }
+      pageInfo {
+        endCursor
+        hasNextPage
+      }
+    }
+  }
+}
+"""
+
+
 def _gh(args: list[str]) -> str | None:
     try:
         return subprocess.check_output(["gh", *args], text=True)
@@ -12,41 +65,42 @@ def _gh(args: list[str]) -> str | None:
         return None
 
 
-JSON_FIELDS = ','.join([
-    "files",
-    "author",
-    "title",
-    "number",
-    "labels",
-    "url"
-])
+def find_prs_without_new_module_tag():
+    hasNextPage = True
+    endCursor = None
+    for _ in tqdm(range(100)):
+        prs = _gh(["api", "graphql", '-F', f'endCursor={'null' if endCursor is None else endCursor}', "-f", f"query={GRAPHQL_QUERY}"])
+        if prs is None:
+            print("gh exited with error")
+            exit(1)
+        data = json.loads(prs)['data']['repository']['pullRequests']
+        endCursor = data['pageInfo']['endCursor']
+        hasNextPage = data['pageInfo']['hasNextPage']
+        prs = [pr['node'] for pr in data['edges']]
+        prs = [{
+            'createdAt': pr['createdAt'],
+            'state': pr['state'],
+            'title': pr['title'],
+            'url': "https://github.com/NixOS/nixpkgs/pull/" + str(pr['number']),
+            'number': pr['number'],
+            'author': pr['author']['login'],
+            'files': [file['node'] for file in pr['files']['edges']],
+            'labels': [label['node']['name'] for label in pr['labels']['edges']],
+        } for pr in prs]
+        prs = [pr for pr in prs if not any(label == '8.has: module (new)' for label in pr['labels'])]
+        prs = [pr for pr in prs if any(file['changeType'] == 'ADDED' and file['path'].startswith('nixos/modules') for file in pr['files'])]
+
+        for pr in reversed(prs):
+            print_pr(pr)
 
 
 def print_pr(pr):
-    print(f'[{pr["number"]}] {pr["author"]["login"]} - {pr["title"]}')
-    print(pr["url"])
-    print(f'Files:')
+    tqdm.write(f'[{pr["state"]}|{pr["number"]}] {pr["author"]} - {pr["title"]}')
+    tqdm.write(pr["url"])
+    tqdm.write(f'Files:')
     for file in pr["files"]:
-        print(f'  {file}')
-
-    print()
-
-
-def find_prs_without_new_module_tag():
-    prs = _gh(["pr", "list", "--json", JSON_FIELDS, "--limit", "50", "--label", "8.has: module (update)"])
-    if prs is None:
-        print("gh exited with error")
-        exit(1)
-    prs = json.loads(prs)
-    for pr in reversed(prs):
-        if any(label["name"] == "8.has: module (new)" for label in pr["labels"]):
-            continue
-
-        if not any(file['path'].startswith("nixos/modules/") and not Path(file['path']).exists() for file in pr['files']):
-            continue
-
-        print_pr(pr)
-
+        tqdm.write(f'  {file}')
+    tqdm.write("")
 
 
 if __name__ == "__main__":