Skip to content

github_api

GitHubAPI

Interact with the GitHub API and perform various git-flow tasks

Source code in all_all_contributors/github_api.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
class GitHubAPI:
    """Interact with the GitHub API and perform various git-flow tasks"""

    def __init__(
        self,
        org_name: str,
        target_repo_name: str,
        github_token: str,
        target_filepath: str = ".all-contributorsrc",
        base_branch: str = "main",
        head_branch: str = "merge-all-contributors",
    ):
        """
        Args:
            org_name (str): The name of the GitHub organisation to target
            target_repo_name (str): The name of the repo within `org_name` that
                will host the combined .all-contributorsrc file
            github_token (str): A GitHub token to authenticate API calls
            target_filepath (str, optional): The filepath within `target_repo_name`
                to the combined `.all-contributorsrc` file.
                (default: ".all-contributorsrc")
            base_branch (str, optional): The name of the default branch in
                `target_repo_name`. (default: "main")
            head_branch (str, optional): A prefix for branches created in
                `target_repo_name` for pull requests.
                (default: "all-all-contributors")
        """
        self.org_name = org_name
        self.target_repo_name = target_repo_name
        self.target_filepath = target_filepath
        self.base_branch = base_branch
        self.head_branch = head_branch

        self.headers = {
            "Accept": "application/vnd.github.v3+json",
            "Authorization": f"token {github_token}",
        }

        self.api_url = "https://api.github.com"

    def create_commit(
        self,
        contents: str,
        commit_msg: str = "Merging all contributors info from across the org",
    ):
        """Create a commit over the GitHub API by creating or updating a file

        Args:
            contents (str): The content of the file to be updated, encoded in base64
            commit_msg (str): A message describing the changes the commit applies.
                (default: "Merging all contributors info from across the org")
        """
        print("Committing changes to file: {}", self.target_filepath)
        url = "/".join(
            [
                self.api_url,
                "repos",
                self.target_repo_name,
                "contents",
                self.target_filepath,
            ]
        )
        body = {
            "message": commit_msg,
            "content": contents,
            "sha": self.sha,
            "branch": self.head_branch,
        }
        put(url, json=body, headers=self.headers)

    def create_ref(self, ref: str, sha: str):
        """Create a new git reference (specifically, a branch) with GitHub's git
        database API endpoint

        Args:
            ref (str): The reference or branch name to create
            sha (str): The SHA of the parent commit to point the new reference to
        """
        print("Creating new branch: {}", ref)
        url = "/".join([self.api_url, "repos", self.target_repo_name, "git", "refs"])
        body = {
            "ref": f"refs/heads/{ref}",
            "sha": sha,
        }
        post_request(url, headers=self.headers, json=body)

    def create_update_pull_request(self):
        """Create or update a Pull Request via the GitHub API"""
        url = "/".join([self.api_url, "repos", self.target_repo_name, "pulls"])
        pr = {
            "title": "Merging all-contributors across the org",
            "body": "",  # FIXME: Add a descriptove PR body here
            "base": self.base_branch,
        }

        if self.pr_exists:
            print("Updating Pull Request...")

            url = "/".join([url, str(self.pr_number)])
            pr["state"] = "open"
            resp = patch_request(
                url,
                headers=self.headers,
                json=pr,
                return_json=True,
            )

            print(f"Pull Request #{resp['number']} updated!")
        else:
            print("Creating Pull Request...")

            pr["head"] = self.head_branch
            resp = post_request(
                url,
                headers=self.headers,
                json=pr,
                return_json=True,
            )

            print(f"Pull Request #{resp['number']} created!")

    def find_existing_pull_request(self):
        """Check if the bot already has an open Pull Request"""
        print("Finding Pull Requests previously opened to merge all contributors files")

        url = "/".join([self.api_url, "repos", self.target_repo_name, "pulls"])
        params = {"state": "open", "sort": "created", "direction": "desc"}
        resp = get_request(url, headers=self.headers, params=params, output="json")

        # Expression to match the head ref
        matches = jmespath.search("[*].head.label", resp)
        indx, match = next(
            (
                (indx, match)
                for (indx, match) in enumerate(matches)
                if self.head_branch in match
            ),
            (None, None),
        )

        if (indx is None) and (match is None):
            print("No relevant Pull Requests found. A new Pull Request will be opened.")
            random_id = "".join(random.sample(string.ascii_letters, 4))
            self.head_branch = "/".join([self.head_branch, random_id])
            self.pr_exists = False
        else:
            print(
                "Relevant Pull Request found. Will push new commits to this Pull Request."
            )

            self.head_branch = match.split(":")[-1]
            self.pr_number = resp[indx]["number"]
            self.pr_exists = True

    def get_ref(self, ref: str) -> dict:
        """Get a git reference (specifically, a HEAD ref) using GitHub's git
        database API endpoint

        Args:
            ref (str): The reference for which to return information for

        Returns:
            dict: The JSON payload response of the request
        """
        print("Pulling info for ref: {}", ref)
        url = "/".join(
            [self.api_url, "repos", self.target_repo_name, "git", "ref", "heads", ref]
        )
        return get_request(url, headers=self.headers, output="json")

    def get_all_repos(self, excluded_repos: list) -> list:
        """
        Get all repositories from a GitHub organization using the GitHub API

        Args:
            excluded_repos (list): A list of excluded repos to skip

        Returns:
            list: A list of remaining repos in the organisation
        """
        org_repos = []

        # First API call
        url = "/".join([self.api_url, "orgs", self.org_name, "repos"])
        params = {"type": "public", "per_page": 100}
        resp = get_request(url, headers=self.headers, params=params)
        for repo in resp.json():
            if repo["name"] not in excluded_repos:
                self.org_repos.append(repo["name"])

        # Paginate over results using the 'link' and rel['next'] parameters from
        # the API response
        # https://docs.github.com/en/rest/using-the-rest-api/using-pagination-in-the-rest-api
        while "link" in resp.headers:
            resp = get_request(
                resp.links["next"]["url"], headers=self.headers, params=params
            )
            for repo in resp.json:
                if repo["name"] not in excluded_repos:
                    self.org_repos.append(repo["name"])

        return org_repos

    def get_contributors_from_repo(
        self, repo: str, filepath=".all-contributorsrc"
    ) -> list:
        """Get contributors from a specific repository using the GitHub API

        Args:
            repo (str): The name of the repository to extract contributors from
            filepath (str): The filepath to extract contributors from (default: .all-contributorsrc)

        Returns:
            list: A list of contributors from the repository
        """
        url = "/".join(
            [self.api_url, "repos", self.org_name, repo, "contents", filepath]
        )
        resp = get_request(url, headers=self.headers, output="json")
        resp = get_request(resp["download_url"], headers=self.headers, output="json")
        return resp["contributors"]

    def get_target_file_contents(self, ref):
        """Download the JSON-formatted contents of a target filepath in a target
        repository inside a target GitHub org.

        Args:
            ref (str): The reference (branch) the file is stored on

        Returns:
            dict: The JSON formatted contents of the target filepath
        """
        url = "/".join(
            [
                self.api_url,
                "repos",
                self.org_name,
                self.target_repo_name,
                "contents",
                self.target_filepath,
            ]
        )
        resp = get_request(
            url, headers=self.headers, params={"ref": ref}, output="json"
        )
        resp = get_ref(resp["download_url"], headers=self.headers, output="json")
        return resp

    def run(self) -> None:
        """Run git flow to make a branch, commit a file, and open a PR"""
        # Check if a PR exists
        github_api.find_existing_pull_request()

        # We want to work against the most up-to-date version of the target file
        if github_api.pr_exists:
            # If a PR exists, pull the file from there
            file_contents = github_api.get_target_file_contents(github_api.head_branch)
        else:
            # Otherwise, pull from the base of the repo
            file_contents = github_api.get_target_file_contents(github_api.base_branch)

        file_contents = inject_config(file_contents, merged_contributors)

        if not github_api.pr_exists:
            # Create a branch to open a PR from
            resp = github_api.get_ref(github_api.base_branch)
            github_api.create_ref(github_api.head_brach, resp["object"]["sha"])

        # base64 encode the updated config file
        encoded_file_contents = yaml.object_to_yaml_str(file_contents).encode("utf-8")
        base64_bytes = base64.b64encode(encoded_file_contents)
        file_contents = base64_bytes.decode("utf-8")

        # Create a commit and open a pull request
        github_api.create_commit(file_contents)
        github_api.create_update_pull_request()

__init__(org_name, target_repo_name, github_token, target_filepath='.all-contributorsrc', base_branch='main', head_branch='merge-all-contributors')

Parameters:

Name Type Description Default
org_name str

The name of the GitHub organisation to target

 required
target_repo_name str

The name of the repo within org_name that will host the combined .all-contributorsrc file

 required
github_token str

A GitHub token to authenticate API calls

 required
target_filepath str

The filepath within target_repo_name to the combined .all-contributorsrc file. (default: ".all-contributorsrc")

 '.all-contributorsrc'
base_branch str

The name of the default branch in target_repo_name. (default: "main")

 'main'
head_branch str

A prefix for branches created in target_repo_name for pull requests. (default: "all-all-contributors")

 'merge-all-contributors'
Source code in all_all_contributors/github_api.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    org_name: str,
    target_repo_name: str,
    github_token: str,
    target_filepath: str = ".all-contributorsrc",
    base_branch: str = "main",
    head_branch: str = "merge-all-contributors",
):
    """
    Args:
        org_name (str): The name of the GitHub organisation to target
        target_repo_name (str): The name of the repo within `org_name` that
            will host the combined .all-contributorsrc file
        github_token (str): A GitHub token to authenticate API calls
        target_filepath (str, optional): The filepath within `target_repo_name`
            to the combined `.all-contributorsrc` file.
            (default: ".all-contributorsrc")
        base_branch (str, optional): The name of the default branch in
            `target_repo_name`. (default: "main")
        head_branch (str, optional): A prefix for branches created in
            `target_repo_name` for pull requests.
            (default: "all-all-contributors")
    """
    self.org_name = org_name
    self.target_repo_name = target_repo_name
    self.target_filepath = target_filepath
    self.base_branch = base_branch
    self.head_branch = head_branch

    self.headers = {
        "Accept": "application/vnd.github.v3+json",
        "Authorization": f"token {github_token}",
    }

    self.api_url = "https://api.github.com"

create_commit(contents, commit_msg='Merging all contributors info from across the org')

Create a commit over the GitHub API by creating or updating a file

Parameters:

Name Type Description Default
contents str

The content of the file to be updated, encoded in base64

 required
commit_msg str

A message describing the changes the commit applies. (default: "Merging all contributors info from across the org")

 'Merging all contributors info from across the org'
Source code in all_all_contributors/github_api.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def create_commit(
    self,
    contents: str,
    commit_msg: str = "Merging all contributors info from across the org",
):
    """Create a commit over the GitHub API by creating or updating a file

    Args:
        contents (str): The content of the file to be updated, encoded in base64
        commit_msg (str): A message describing the changes the commit applies.
            (default: "Merging all contributors info from across the org")
    """
    print("Committing changes to file: {}", self.target_filepath)
    url = "/".join(
        [
            self.api_url,
            "repos",
            self.target_repo_name,
            "contents",
            self.target_filepath,
        ]
    )
    body = {
        "message": commit_msg,
        "content": contents,
        "sha": self.sha,
        "branch": self.head_branch,
    }
    put(url, json=body, headers=self.headers)

create_ref(ref, sha)

Create a new git reference (specifically, a branch) with GitHub's git database API endpoint

Parameters:

Name Type Description Default
ref str

The reference or branch name to create

 required
sha str

The SHA of the parent commit to point the new reference to

 required
Source code in all_all_contributors/github_api.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def create_ref(self, ref: str, sha: str):
    """Create a new git reference (specifically, a branch) with GitHub's git
    database API endpoint

    Args:
        ref (str): The reference or branch name to create
        sha (str): The SHA of the parent commit to point the new reference to
    """
    print("Creating new branch: {}", ref)
    url = "/".join([self.api_url, "repos", self.target_repo_name, "git", "refs"])
    body = {
        "ref": f"refs/heads/{ref}",
        "sha": sha,
    }
    post_request(url, headers=self.headers, json=body)

create_update_pull_request()

Create or update a Pull Request via the GitHub API

Source code in all_all_contributors/github_api.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def create_update_pull_request(self):
    """Create or update a Pull Request via the GitHub API"""
    url = "/".join([self.api_url, "repos", self.target_repo_name, "pulls"])
    pr = {
        "title": "Merging all-contributors across the org",
        "body": "",  # FIXME: Add a descriptove PR body here
        "base": self.base_branch,
    }

    if self.pr_exists:
        print("Updating Pull Request...")

        url = "/".join([url, str(self.pr_number)])
        pr["state"] = "open"
        resp = patch_request(
            url,
            headers=self.headers,
            json=pr,
            return_json=True,
        )

        print(f"Pull Request #{resp['number']} updated!")
    else:
        print("Creating Pull Request...")

        pr["head"] = self.head_branch
        resp = post_request(
            url,
            headers=self.headers,
            json=pr,
            return_json=True,
        )

        print(f"Pull Request #{resp['number']} created!")

find_existing_pull_request()

Check if the bot already has an open Pull Request

Source code in all_all_contributors/github_api.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def find_existing_pull_request(self):
    """Check if the bot already has an open Pull Request"""
    print("Finding Pull Requests previously opened to merge all contributors files")

    url = "/".join([self.api_url, "repos", self.target_repo_name, "pulls"])
    params = {"state": "open", "sort": "created", "direction": "desc"}
    resp = get_request(url, headers=self.headers, params=params, output="json")

    # Expression to match the head ref
    matches = jmespath.search("[*].head.label", resp)
    indx, match = next(
        (
            (indx, match)
            for (indx, match) in enumerate(matches)
            if self.head_branch in match
        ),
        (None, None),
    )

    if (indx is None) and (match is None):
        print("No relevant Pull Requests found. A new Pull Request will be opened.")
        random_id = "".join(random.sample(string.ascii_letters, 4))
        self.head_branch = "/".join([self.head_branch, random_id])
        self.pr_exists = False
    else:
        print(
            "Relevant Pull Request found. Will push new commits to this Pull Request."
        )

        self.head_branch = match.split(":")[-1]
        self.pr_number = resp[indx]["number"]
        self.pr_exists = True

get_all_repos(excluded_repos)

Get all repositories from a GitHub organization using the GitHub API

Parameters:

Name Type Description Default
excluded_repos list

A list of excluded repos to skip

 required

Returns:

Name Type Description
list list

A list of remaining repos in the organisation
Source code in all_all_contributors/github_api.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def get_all_repos(self, excluded_repos: list) -> list:
    """
    Get all repositories from a GitHub organization using the GitHub API

    Args:
        excluded_repos (list): A list of excluded repos to skip

    Returns:
        list: A list of remaining repos in the organisation
    """
    org_repos = []

    # First API call
    url = "/".join([self.api_url, "orgs", self.org_name, "repos"])
    params = {"type": "public", "per_page": 100}
    resp = get_request(url, headers=self.headers, params=params)
    for repo in resp.json():
        if repo["name"] not in excluded_repos:
            self.org_repos.append(repo["name"])

    # Paginate over results using the 'link' and rel['next'] parameters from
    # the API response
    # https://docs.github.com/en/rest/using-the-rest-api/using-pagination-in-the-rest-api
    while "link" in resp.headers:
        resp = get_request(
            resp.links["next"]["url"], headers=self.headers, params=params
        )
        for repo in resp.json:
            if repo["name"] not in excluded_repos:
                self.org_repos.append(repo["name"])

    return org_repos

get_contributors_from_repo(repo, filepath='.all-contributorsrc')

Get contributors from a specific repository using the GitHub API

Parameters:

Name Type Description Default
repo str

The name of the repository to extract contributors from

 required
filepath str

The filepath to extract contributors from (default: .all-contributorsrc)

 '.all-contributorsrc'

Returns:

Name Type Description
list list

A list of contributors from the repository
Source code in all_all_contributors/github_api.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def get_contributors_from_repo(
    self, repo: str, filepath=".all-contributorsrc"
) -> list:
    """Get contributors from a specific repository using the GitHub API

    Args:
        repo (str): The name of the repository to extract contributors from
        filepath (str): The filepath to extract contributors from (default: .all-contributorsrc)

    Returns:
        list: A list of contributors from the repository
    """
    url = "/".join(
        [self.api_url, "repos", self.org_name, repo, "contents", filepath]
    )
    resp = get_request(url, headers=self.headers, output="json")
    resp = get_request(resp["download_url"], headers=self.headers, output="json")
    return resp["contributors"]

get_ref(ref)

Get a git reference (specifically, a HEAD ref) using GitHub's git database API endpoint

Parameters:

Name Type Description Default
ref str

The reference for which to return information for

 required

Returns:

Name Type Description
dict dict

The JSON payload response of the request
Source code in all_all_contributors/github_api.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def get_ref(self, ref: str) -> dict:
    """Get a git reference (specifically, a HEAD ref) using GitHub's git
    database API endpoint

    Args:
        ref (str): The reference for which to return information for

    Returns:
        dict: The JSON payload response of the request
    """
    print("Pulling info for ref: {}", ref)
    url = "/".join(
        [self.api_url, "repos", self.target_repo_name, "git", "ref", "heads", ref]
    )
    return get_request(url, headers=self.headers, output="json")

get_target_file_contents(ref)

Download the JSON-formatted contents of a target filepath in a target repository inside a target GitHub org.

Parameters:

Name Type Description Default
ref str

The reference (branch) the file is stored on

 required

Returns:

Name Type Description
dict

The JSON formatted contents of the target filepath
Source code in all_all_contributors/github_api.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def get_target_file_contents(self, ref):
    """Download the JSON-formatted contents of a target filepath in a target
    repository inside a target GitHub org.

    Args:
        ref (str): The reference (branch) the file is stored on

    Returns:
        dict: The JSON formatted contents of the target filepath
    """
    url = "/".join(
        [
            self.api_url,
            "repos",
            self.org_name,
            self.target_repo_name,
            "contents",
            self.target_filepath,
        ]
    )
    resp = get_request(
        url, headers=self.headers, params={"ref": ref}, output="json"
    )
    resp = get_ref(resp["download_url"], headers=self.headers, output="json")
    return resp

run()

Run git flow to make a branch, commit a file, and open a PR

Source code in all_all_contributors/github_api.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def run(self) -> None:
    """Run git flow to make a branch, commit a file, and open a PR"""
    # Check if a PR exists
    github_api.find_existing_pull_request()

    # We want to work against the most up-to-date version of the target file
    if github_api.pr_exists:
        # If a PR exists, pull the file from there
        file_contents = github_api.get_target_file_contents(github_api.head_branch)
    else:
        # Otherwise, pull from the base of the repo
        file_contents = github_api.get_target_file_contents(github_api.base_branch)

    file_contents = inject_config(file_contents, merged_contributors)

    if not github_api.pr_exists:
        # Create a branch to open a PR from
        resp = github_api.get_ref(github_api.base_branch)
        github_api.create_ref(github_api.head_brach, resp["object"]["sha"])

    # base64 encode the updated config file
    encoded_file_contents = yaml.object_to_yaml_str(file_contents).encode("utf-8")
    base64_bytes = base64.b64encode(encoded_file_contents)
    file_contents = base64_bytes.decode("utf-8")

    # Create a commit and open a pull request
    github_api.create_commit(file_contents)
    github_api.create_update_pull_request()