import example_ardoq_client import argparse import json import urllib.request import urllib.parse import pprint import time parser = argparse.ArgumentParser() parser.add_argument( "--host", metavar="ARDOQ_API_HOST", type=str, help="Required if you are using a custom domain", ) parser.add_argument( "--org", metavar="ARDOQ_ORG_LABEL", type=str, help="The label associated with your org", ) parser.add_argument( "--token", metavar="ARDOQ_API_TOKEN", type=str, help="Your secret API token" ) parser.add_argument( "--github-token", metavar="GITHUB_TOKEN", type=str, help="Optional Github API token" ) parser.add_argument( "github_org", metavar="GITHUB_ORG", type=str, help="Name of your Github org" ) parser.add_argument( "workspace", metavar="WORKSPACE", type=str, help="Ardoq identifier of your workspace", ) args = parser.parse_args() ### print("Using Workspace: {}".format(args.workspace)) print("Using Github Org: {}".format(args.github_org)) def repos(): req = urllib.request.Request( "https://api.github.com/orgs/{}/repos".format(args.github_org) ) if args.github_token: req.add_header("Authorization", "Bearer {}".format(args.github_token)) with urllib.request.urlopen(req) as resp: repos = json.loads(resp.read().decode("utf-8")) return list(filter(lambda repo: not repo["fork"], repos)) def contributors(repo_name): req = urllib.request.Request( "https://api.github.com/repos/{}/{}/contributors".format( args.github_org, repo_name ) ) if args.github_token: req.add_header("Authorization", "Bearer {}".format(args.github_token)) with urllib.request.urlopen(req) as resp: if resp.getcode() == 200: return json.loads(resp.read().decode("utf-8")) return [] api = example_ardoq_client.API( ardoq_api_host=args.host, ardoq_org_label=args.org, ardoq_api_token=args.token ) github_workspace = args.workspace ctx = api.read_workspace_context(github_workspace) comp_types = {t["name"]: t["typeId"] for t in ctx["componentTypes"]} ref_types = {t["name"]: t["type"] for t in ctx["referenceTypes"]} repo_components = api.list_components( {"rootWorkspace": github_workspace, "typeId": comp_types["GithubRepo"]} ) repo_lookup = {c["name"]: c for c in repo_components} user_components = api.list_components( {"rootWorkspace": github_workspace, "typeId": comp_types["GithubUser"]} ) user_id_lookup = {c["name"]: c["_id"] for c in user_components} contrib_references = api.list_references( {"rootWorkspace": github_workspace, "type": ref_types["Contributed to"]} ) contrib_lookup = {(r["source"], r["target"]): r for r in contrib_references} batch = example_ardoq_client.Batch() org_repos = repos() total_org_repos = len(org_repos) for index, repo in enumerate(org_repos, 1): if not args.github_token: time.sleep(1) repo_name = repo["name"] qualified_repo_name = args.github_org + "/" + repo_name print("{}/{} : {}".format(index, total_org_repos, qualified_repo_name)) repo_component = repo_lookup.get(qualified_repo_name) if repo_component: repo_id = repo_component["_id"] if repo["description"] != repo_component["description"]: batch.update_component(repo_id, {"description": repo["description"]}) else: repo_id = qualified_repo_name batch.create_component( { "rootWorkspace": github_workspace, "typeId": comp_types["GithubRepo"], "name": qualified_repo_name, "description": repo["description"], }, batchId=repo_id, ) for user in contributors(repo_name): user_name = user["login"] user_id = user_id_lookup.get(user_name) if user_id is None: user_id = "user-" + user_name user_id_lookup[user_name] = user_id batch.create_component( { "rootWorkspace": github_workspace, "typeId": comp_types["GithubUser"], "name": user_name, }, batchId=user_id, ) contrib_reference = contrib_lookup.get((user_id, repo_id)) n = user["contributions"] displayText = "{} commit{}".format(n, "s" if n > 1 else "") if contrib_reference: if contrib_reference.get("customFields", {}).get("contributions") != n: batch.update_reference( contrib_reference["_id"], { "displayText": displayText, "customFields": {"contributions": n}, }, ) else: batch.create_reference( { "source": user_id, "target": repo_id, "type": ref_types["Contributed to"], "displayText": displayText, "customFields": {"contributions": n}, } ) print("Payload -------------------------------------") print(json.dumps(batch.body)) print("---------------------------------------------") if batch.is_empty(): print("No change detected... nothing to do") else: resp = api.batch(batch.body) pprint.pprint(resp)