github GraphQL access

github

#!/usr/bin/env python

import requests
import json
import os
import sys
import argparse


ENTRYPOINT = "https://api.github.com/graphql"


def __get_headers(tok_file):
    with open(tok_file) as fp: token = fp.read()
    token = token.rstrip()
    headers = {
        "Authorization" : "token %s" % token
    }
    return headers


def __get_owner_info(repo):
    toks = repo.replace("https://github.com/", "")
    toks = toks.split("/")
    return toks[0], toks[1]


def __get_owner_pr_info(repo):
    toks = repo.replace("https://github.com/", "")
    toks = toks.split("/")
    return toks[0], toks[1], toks[3]


def __get_response(payload, tok_file):
    headers = __get_headers(tok_file)
    data = json.dumps(payload)
    res = requests.post(ENTRYPOINT, data=data, headers=headers)
    return json.loads(res.content)


def __pretty_print(data, indent):
    print(json.dumps(data, indent=indent))


def __base_query(content, tok_file):
    payload = {
        "query" : "query { " + content + " }"
    }
    return __get_response(payload, tok_file)


def __core_repo_query(repo_owner, repo_name, content, tok_file):
    content = "repository(owner:\"%s\", name: \"%s\") { %s }" % \
        (repo_owner, repo_name, content)
    return __base_query(content, tok_file)


def __base_repo_query(repo, content, tok_file):
    repo_owner, repo_name = __get_owner_info(repo)
    return __core_repo_query(repo_owner, repo_name, content, tok_file)


def list_all_types(args):
    content = """
__schema {
  types {
    name
    kind
    description
    fields {
      name
    }
  }
}"""
    return __base_query(content, args.token)


def list_type(args):
    content = """
__type(name: "%s") {
  kind
  description
  enumValues {
    name
    description
  }
  inputFields {
    name
    description
  }
  interfaces {
    name
    description
  }
  ofType {
    name
    description
  }
  possibleTypes {
    name
    description
  }
  fields {
    name
    description
    type {
      name
      kind
    }
  }
}""" % args.type
    return __base_query(content, args.token)


def default_branch(args):
    return __base_repo_query(args.repo, "defaultBranchRef { name }", args.token)


PR_INFO = """
  title
  number
  id
  author { login }
  commits(last:1) {
    nodes {
      commit {
        oid
        status { state }
      }
    }
  }
  baseRefName
  headRefName
"""


def open_prs(args):
    content = """
pullRequests(states:[OPEN], first:40%s) {
  pageInfo {
    hasNextPage
    endCursor
  }
  edges { node {
    %s
  } }
}
"""
    page_info = {
        "hasNextPage" : True,
        "endCursor" : ""
    }
    ret = {}
    while page_info["hasNextPage"]:
        if page_info["endCursor"]:
            txt = ", after:\"%s\"" % page_info["endCursor"]
        else:
            txt = ""
        res = __base_repo_query(args.repo, content % (txt, PR_INFO), args.token)
        page_info = res["data"]["repository"]["pullRequests"]["pageInfo"]
        del res["data"]["repository"]["pullRequests"]["pageInfo"]
        ret.update(res)
    return ret


def open_prs_by(args):
    content = """
search(query:"repo:%s/%s type:pr is:open author:%s", type:ISSUE, first:40%s) {
  pageInfo {
    hasNextPage
    endCursor
  }
  nodes {
    ... on PullRequest {
      %s
    }
  }
}
"""
    page_info = {
        "hasNextPage" : True,
        "endCursor" : ""
    }
    ret = {}
    repo_owner, repo_name = __get_owner_info(args.repo)
    while page_info["hasNextPage"]:
        if page_info["endCursor"]:
            txt = ", after:\"%s\"" % page_info["endCursor"]
        else:
            txt = ""
        searchTxt = content % (repo_owner, repo_name, args.user, txt, PR_INFO)
        res = __base_query(searchTxt, args.token)
        page_info = res["data"]["search"]["pageInfo"]
        del res["data"]["search"]["pageInfo"]
        ret.update(res)
    return ret


def list_collaborators(args):
    content = """
collaborators(affiliation:ALL) {
  edges {
    node {
      name
      id
      login
    }
  }
}"""
    return __base_repo_query(args.repo, content, args.token)


def organization(args):
    content = """
organization(login:"%s") {
  description
  id
  login
  name
  repositories(first:40%s) {
    pageInfo {
      hasNextPage
      endCursor
    }
    edges { node {
      description
      homepageUrl
      id
      isPrivate
      name
      url
    } }
  }
  url
  websiteUrl
}
"""
    page_info = {
        "hasNextPage" : True,
        "endCursor" : ""
    }
    ret = None
    while page_info["hasNextPage"]:
        if page_info["endCursor"]:
            txt = ", after:\"%s\"" % page_info["endCursor"]
        else:
            txt = ""
        res = __base_query(content % (args.org, txt), args.token)
        page_info = res["data"]["organization"]["repositories"]["pageInfo"]
        del res["data"]["organization"]["repositories"]["pageInfo"]
        if ret is None:
            ret = res
        else:
            ret["data"]["organization"]["repositories"]["edges"] += res["data"]["organization"]["repositories"]["edges"]
    return ret


def pr_info(args):
    repo_owner, repo_name, pr_id = __get_owner_pr_info(args.pr)
    content = """
pullRequest(number:%s) {
  %s
}
""" % (pr_id, PR_INFO)
    return __core_repo_query(repo_owner, repo_name, content, args.token)


def query(args):
    return __base_query(args.content, args.token)


def user_info(args):
    header = "viewer {"
    if args.user:
        header = "user(login:\"%s\") {" % args.user
    content = """
%s
  login
  email
  name
  company
  id
  location
  websiteUrl
}""" % header
    return __base_query(content, args.token)


def main():
    # main parser
    parser = argparse.ArgumentParser()
    parser.add_argument("-token", type=str,
                        default=os.path.join(os.path.expanduser("~"),
                                             ".github-api-token"),
                        help="Path to the token file for authentication")
    parser.add_argument("-indent", type=int, default=2,
                        help="Pretty print indent width")
    subs = parser.add_subparsers()
    # default_branch
    args_default_branch = subs.add_parser(
        "default_branch", description="Default branch of the repo")
    args_default_branch.add_argument("repo", default=None, help="Link to the repo")
    args_default_branch.set_defaults(func=default_branch)
    # list_all_types
    args_list_all_types = subs.add_parser(
        "list_all_types",
        description="List all supported types in github's gql API")
    args_list_all_types.set_defaults(func=list_all_types)
    # list_collaborators
    args_list_collaborators = subs.add_parser(
        "list_collaborators", description="List collaborators for a given repo")
    args_list_collaborators.add_argument("repo", default=None, help="Link to the repo")
    args_list_collaborators.set_defaults(func=list_collaborators)
    # list_type
    args_list_type = subs.add_parser(
        "list_type",
        description="List info about the given type in github's gql API")
    args_list_type.add_argument(
        "type", default=None, help="The type for which info is needed")
    args_list_type.set_defaults(func=list_type)
    # open_prs
    args_open_prs = subs.add_parser(
        "open_prs", description="List of open PRs against the repo")
    args_open_prs.add_argument("repo", default=None, help="Link to the repo")
    args_open_prs.set_defaults(func=open_prs)
    # open_prs_by
    args_open_prs_by = subs.add_parser(
        "open_prs_by", description="List open PRs against the repo by a user")
    args_open_prs_by.add_argument("repo", default=None, help="Link to the repo")
    args_open_prs_by.add_argument("user", default=None, help="The user")
    args_open_prs_by.set_defaults(func=open_prs_by)
    # org_info
    args_org_info = subs.add_parser("org_info", description="List organization info")
    args_org_info.add_argument("org", default=None, nargs="?",
                               help="The organization whose info is needed")
    args_org_info.set_defaults(func=organization)
    # pr_info
    args_pr_info = subs.add_parser("pr_info", description="Get info about a PR")
    args_pr_info.add_argument("pr", default=None, help="Link to the PR")
    args_pr_info.set_defaults(func=pr_info)
    # query
    args_query = subs.add_parser("query", description="Raw query")
    args_query.add_argument("content", default=None, help="Raw query string")
    args_query.set_defaults(func=query)
    # user_info
    args_user_info = subs.add_parser("user_info", description="List user info")
    args_user_info.add_argument("user", default=None, nargs="?",
                                help="The user whose info is needed")
    args_user_info.set_defaults(func=user_info)
    # run
    args = parser.parse_args()
    __pretty_print(args.func(args), args.indent)


if __name__ == "__main__":
    main()