Transfer issues from pytorch/torchdynamo to pytorch/pytorch

gh_transfer_issues.py

#!/usr/bin/env python3

from collections import namedtuple
from typing import Tuple, List


GH_GET_REPO_ID = """
query ($owner: String = "pytorch", $name: String = "pytorch") {
  repository(name: $name, owner: $owner) {
    id
  }
}
"""

GH_GET_ISSUES_INFO_FRAGMENT = """
fragment IssueInfo on IssueConnection {
  nodes {
    id
    url
    title
  }
  pageInfo {
    hasPreviousPage
    startCursor
  }
}
"""

GHIssueInfo = namedtuple("GHIssueInfo", ["id", "url", "title"])

GH_GET_OPEN_ISSUES = GH_GET_ISSUES_INFO_FRAGMENT + """
query ($owner: String = "pytorch", $name: String = "torchdynamo") {
  repository(name: $name, owner: $owner) {
    issues(last: 100, states: OPEN) {
      ...IssueInfo
    }
  }
}
"""

GH_TRANSFER_ISSUE = """
mutation ($repository: ID!, $issue: ID!) {
  transferIssue(input: {issueId: $issue, repositoryId: $repository}) {
    issue {
      number
      url
    }
  }
}
"""


def gh_graphql(query, **kwargs):
    import os
    import json
    from urllib.request import urlopen, Request
    token = os.getenv("GITHUB_TOKEN")
    headers = {"Accept": "application/vnd.github.v3+json", "Authorization": f"token {token}"}

    data = json.dumps({"query": query, "variables": kwargs}).encode()
    with urlopen(Request("https://api.github.com/graphql", data=data, headers=headers)) as conn:
        return json.load(conn)


def gh_get_repository_id(owner: str = "pytorch", name: str = "pytorch") -> str:
    return gh_graphql(GH_GET_REPO_ID, owner=owner, name=name)["data"]["repository"]["id"]


def gh_get_open_issues(owner: str = "pytorch", name: str = "torchdynamo") -> Tuple[List[GHIssueInfo], bool]:
    issues = gh_graphql(GH_GET_OPEN_ISSUES, owner=owner, name=name)["data"]["repository"]["issues"]
    rc: List[GHIssueInfo] = []
    for issue in issues["nodes"]:
        rc.append(GHIssueInfo(id=issue["id"], url=issue["url"], title=issue["title"]))
    return rc, bool(issues["pageInfo"]["hasPreviousPage"])


def gh_transfer_issue(issue_id: str, repo_id: str) -> str:
    rc = gh_graphql(GH_TRANSFER_ISSUE, repository=repo_id, issue=issue_id)
    return rc["data"]["transferIssue"]["issue"]["url"]


def main() -> None:
    tgt_repo_id = gh_get_repository_id()
    issues, has_more = gh_get_open_issues()
    for issue in issues:
        new_url = gh_transfer_issue(issue.id, tgt_repo_id)
        print(f"{issue.url} -> {new_url}")


if __name__ == "__main__":
    main()