github_project.py

import requests
import sys
import json
from pprint import pprint
from typing import Optional, Dict, Union, Tuple, Any, List
import argparse

PROJECT_FIELDS = """
query($org: String!, $number: Int!) {
  organization(login: $org) {
    projectV2(number: $number) {
      id
      fields(first:20) {
        nodes {
          ... on ProjectV2Field {
            id
            name
          }
          ... on ProjectV2SingleSelectField {
            id
            name
            options {
              id
              name
            }
          }
        }
      }
    }
  }
}
"""

ISSUE_ID = """
query($org: String!, $repo: String!, $number: Int!) {
  organization(login: $org) {
    repository(name: $repo) {
      issue(number: $number) {
        id
      }
    }
  }
}
"""

SEARCH_ISSUES = """
query($searchQuery: String!) {
  search(
    query: $searchQuery
    type: ISSUE
    first: 100
  ) {
    edges {
      node {
        ... on Issue {
          id
          number
          title
          url
        }
      }
    }
  }
}
"""

ADD_ISSUE_TO_PROJECT = """
mutation AddToProject($project: ID!, $issue: ID!) {
  addProjectNextItem(input: { projectId: $project, contentId: $issue }) {
    projectNextItem {
      id
    }
  }
}
"""

ALL_ISSUES_IN_PROJECT = """
query($org: String!, $number: Int!) {
  organization(login: $org) {
    projectV2(number: $number) {
      items(last: 100) {
        nodes {
          id

          fieldValues(first: 8) {
            nodes {
              ... on ProjectV2ItemFieldSingleSelectValue {
                name
                field {
                  ... on ProjectV2FieldCommon {
                    name
                  }
                }
              }
              ... on ProjectV2ItemFieldTextValue {
                text
                field {
                  ... on ProjectV2FieldCommon {
                    name
                  }
                }
              }
              ... on ProjectV2ItemFieldPullRequestValue {
                pullRequests(last: 10) {
                  nodes {
                    url
                    number
                  }
                }
              }
            }
          }

          content {
            ...on Issue {
              title
              number
              url
              state
              milestone {
                number
              }
            }
          }
        }
      }
    }
  }
}
"""

SET_PROJECT_STATUS = """
mutation UpdateProject($project: ID!, $item: ID!, $status_field: ID!, $status_value: String!) {
  updateProjectV2ItemFieldValue(input: { projectId: $project, itemId: $item, fieldId: $status_field, value: { singleSelectOptionId: $status_value } }) {
    projectV2Item {
      id
    }
  }
}
"""

session = requests.session()

def graphql(query: str, variables: Dict[str, Union[int, str]]) -> Dict[Any, Any]:
  req = session.post("https://api.github.com/graphql", json={"query": query, "variables": variables})
  req.raise_for_status()
  jsondata = req.json()

  if "errors" in jsondata:
    print("Something went wrong:")
    pprint(jsondata["errors"])
    raise RuntimeError

  if not "data" in jsondata:
    print("No data in return - instead we got:")
    pprint(jsondata)
    raise RuntimeError

  return jsondata

def gv(data_dict: Dict, value_key: str, def_value:Any = None) -> Any:
  keys = value_key.split(".")

  cd = data_dict
  for key in keys:
    if isinstance(cd, str) and cd != "null":
      cd = json.loads(cd)
    if key in cd:
      cd = cd[key]
    else:
      return def_value
  return cd

def project_id_and_fields(project_id: int) -> Tuple[str, Dict]:
  data = graphql(PROJECT_FIELDS, {"org": "llvm", "number": project_id})
  fields = {}
  for node in gv(data, "data.organization.projectV2.fields.nodes", []):
    fields[node["name"]] = {
      "id": node["id"],
      "options": {n['name']:n['id'] for n in gv(node, "options", [])}
    }
  return (gv(data, "data.organization.projectV2.id", 0), fields)

def issue_id(issue_id: int) -> str:
  data = graphql(ISSUE_ID, {"org": "llvm", "repo": "llvm-project", "number": issue_id})
  return gv(data, "data.organization.repository.issue.id", 0)

def search_issues(query: str) -> List[Dict[str, Any]]:
  data = graphql(SEARCH_ISSUES, {"searchQuery": query})
  issues = []
  for node in gv(data, "data.search.edges", []):
    node = node["node"]
    issues.append({
      "id": node["id"],
      "number": node["number"],
      "title": node["title"],
      "url": node["url"]
    })
  return issues

def add_issue_to_project(project_id:str, issue_id:str) -> str:
  data = graphql(ADD_ISSUE_TO_PROJECT, {"project": project_id, "issue": issue_id})
  return gv(data, "data.addProjectNextItem.projectNextItem.id", None)

def get_issues_from_project(org: str, project_id: int) -> List[Dict[str, Any]]:
  data = graphql(ALL_ISSUES_IN_PROJECT, {"org": org, "number": project_id})
  issues = []
  for node in gv(data, "data.organization.projectV2.items.nodes", []):
    idata = {
      "milestone": gv(node, "content.milestone.number", 0),
      "number": gv(node, "content.number", 0),
      "title": gv(node, "content.title", ""),
      "state": gv(node, "content.state", ""),
      "id": gv(node, "id", None),
      "pr_url": [],
      "pr": [],
      "url": gv(node, "content.url", None),
      "status": None,
    }

    for field in gv(node, "fieldValues.nodes", []):
      if "pullRequests" in field:
        for pr in gv(field, "pullRequests.nodes", []):
          idata["pr_url"].append(pr["url"])
          idata["pr"].append(str(pr["number"]))

      if "field" in field and gv(field, "field.name") == "Status":
        idata["status"] = field["name"]

    issues.append(idata)

  return issues

def set_project_status(project_id:str, issue_id:str, status_id:str, new_status:str) -> str:
  # $project: ID!, $item: ID!, $status_field: ID!, $status_value: String!
  data = graphql(SET_PROJECT_STATUS, {"project": project_id, "item": issue_id, "status_field": status_id, "status_value": new_status})
  return gv(data, "data.updateProjectNxetItemField.projectNextItem.id", "")

def list_command(args: Any):

  project_id = None
  status_to_set = None
  status_id = None
  if args.set_status:
    project_id, fields = project_id_and_fields(args.projectnum)
    status_id = gv(fields, "Status.id", None)
    possible_status = gv(fields, "Status.options")
    if not args.set_status in possible_status:
      print(f"Can't set status to {args.set_status} - possible values are: {', '.join(possible_status.keys())}")
      sys.exit(1)

    status_to_set = possible_status[args.set_status]

  issues = get_issues_from_project(args.org, args.projectnum)

  filtered_issues = issues
  if args.milestone or args.status or args.has_pr:
    if args.milestone:
      filtered_issues = [iss for iss in filtered_issues if iss['milestone'] == args.milestone]
    if args.status:
      filtered_issues = [iss for iss in filtered_issues if iss['status'] == args.status]
    if args.state:
      filtered_issues = [iss for iss in filtered_issues if iss['state'] == args.state]
    if args.has_pr:
      filtered_issues = [iss for iss in filtered_issues if len(iss['pr']) > 0]

  for iss in filtered_issues:
    if args.output:
      stuff = args.output.split(",")
      output = []
      for s in stuff:
        val = iss[s]
        if isinstance(val, list):
          output.append("|".join(val))
        else:
          output.append(str(val))

      print(";".join(output))
    else:
      pr = ', '.join([p for p in iss["pr_url"]])
      if pr:
        pr = f"  PR: {pr}\n"
      print(f"{iss['number']}: {iss['title']}\n  State: {iss['state']}\n  URL: {iss['url']}\n  milestone: {iss['milestone']}\n{pr}  Status: {iss['status']}")

    if project_id and status_to_set and status_id:
      print(f"Setting status to {args.set_status} ...\n")
      set_project_status(project_id, iss['id'], status_id, status_to_set)

def search_command(args: Any):
  issues = search_issues(args.query)
  project_id = None
  for iss in issues:
    if args.add_to_project:
      if not project_id:
        project_id, _ = project_id_and_fields(args.projectnum)
      print(f"Adding {iss['number']} to project {args.projectnum}...")
      add_issue_to_project(project_id, iss["number"])
    else:
      print(f"{iss['number']}: {iss['title']}\n  URL: {iss['url']}")

def search_add_command(args: Any):
  issues = search_issues(args.query)
  project_id, _ = project_id_and_fields(args.projectnum)
  for iss in issues:
    print(f"Addding {iss['number']} to {args.projectnum} ({project_id})")
    add_issue_to_project(project_id, iss['id'])

if __name__ == '__main__':

  parser = argparse.ArgumentParser()
  parser.add_argument("--token", required=True)
  parser.add_argument("--org", default="llvm")
  parser.add_argument("--projectnum", default=3)

  commands = parser.add_subparsers(dest="command")

  listparser = commands.add_parser("list")
  listparser.add_argument("--milestone", default=None, type=int)
  listparser.add_argument("--status", default=None, type=str)
  listparser.add_argument("--output", default=None, type=str)
  listparser.add_argument("--has-pr", default=False, action="store_true")
  listparser.add_argument("--state", default=None, type=str)
  listparser.add_argument("--set-status", default=None, type=str)

  searchparser = commands.add_parser("search")
  searchparser.add_argument("--query", required=True)
  searchparser.add_argument("--add-to-project", default=False, action="store_true")

  args = parser.parse_args()

  session.headers["Authorization"] = "Bearer " + args.token

  if args.command == "list":
    list_command(args)
  elif args.command == "search":
    search_command(args)