Skip to content

all_all_contributors

cli

get_github_token()

Read a GitHub token from the environment

Source code in all_all_contributors/cli.py 
14
15
16
17
18
19
20
def get_github_token() -> str | None:
    """Read a GitHub token from the environment"""
    token = getenv("INPUT_GITHUB_TOKEN")
    if token is None:
        print("Environment variable INPUT_GITHUB_TOKEN is not defined")
        raise typer.Exit(code=1)
    return token

load_excluded_repos()

Load excluded repositories from a file

Returns:

Name Type Description
set set

A set of excluded repository names
Source code in all_all_contributors/cli.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def load_excluded_repos() -> set:
    """Load excluded repositories from a file

    Returns:
        set: A set of excluded repository names
    """
    ignore_file = getenv("INPUT_IGNORE_FILE", ".repoignore")
    if path.exists(ignore_file):
        with open(ignore_file) as f:
            excluded = filter(lambda line: not line.startswith("#"), f.readlines())
    else:
        print(f"[skipping] No file found: {ignore_file}.")
        excluded = []

    return set(excluded)

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()

http_requests

A set of basic functional wrappers around HTTP verbs

  • get: customise response format
  • patch
  • post

get_request(url, headers={}, params={}, output='default')

Send a GET request to an HTTP API endpoint

Parameters:

Name Type Description Default
url str

The URL to send the request to

 required
headers dict

A dictionary of headers to send with the request. Defaults to an empty dict.

 {}
params dict

A dictionary of parameters to send with the request. Defaults to an empty dict.

 {}
output str

The format in which to output the response in. Currently accepts 'default', 'json' or 'text'. 'default' does not apply any format parsing of the response.

 'default'

Returns:

Name Type Description
resp

A formatted response from the HTTP request

Raises:

Type Description
ValueError

An unexpected output format was chosen.
Source code in all_all_contributors/http_requests.py 
12
13
14
15
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
def get_request(url, headers={}, params={}, output="default"):
    """Send a GET request to an HTTP API endpoint

    Args:
        url (str): The URL to send the request to
        headers (dict, optional): A dictionary of headers to send with the
            request. Defaults to an empty dict.
        params (dict, optional): A dictionary of parameters to send with the
            request. Defaults to an empty dict.
        output (str): The format in which to output the response in. Currently
            accepts 'default', 'json' or 'text'. 'default' does not apply any
            format parsing of the response.

    Returns:
        resp: A formatted response from the HTTP request

    Raises:
        ValueError: An unexpected output format was chosen.
    """
    accepted_formats = ["default", "json", "text"]
    if output not in accepted_formats:
        raise ValueError(
            "Invalid output format. Please choose one of the following options: %s"
            % accepted_formats
        )

    resp = requests.get(url, headers=headers, params=params)

    if not resp:
        raise requests.HTTPError(f"{resp.text}\nRequest URL: {url}")

    if output == "default":
        return resp
    elif output == "json":
        return resp.json()
    elif output == "text":
        return resp.text

patch_request(url, headers={}, json={}, return_json=False)

Send a PATCH request to an HTTP API endpoint

Parameters:

Name Type Description Default
url str

The URL to send the request to

 required
headers dict

A dictionary of any headers to send with the request. Defaults to an empty dictionary.

 {}
json dict

A dictionary containing JSON payload to send with the request. Defaults to an empty dictionary.

 {}
return_json bool

Return the JSON payload response. Defaults to False.

 False

Returns:

Name Type Description
resp

A JSON formatted response from a HTTP response

Raises:

Type Description
HTTPError

An error was returned by the HTTP request
Source code in all_all_contributors/http_requests.py 
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
def patch_request(url, headers={}, json={}, return_json=False):
    """Send a PATCH request to an HTTP API endpoint

    Args:
        url (str): The URL to send the request to
        headers (dict, optional): A dictionary of any headers to send with the
            request. Defaults to an empty dictionary.
        json (dict, optional): A dictionary containing JSON payload to send with
            the request. Defaults to an empty dictionary.
        return_json (bool, optional): Return the JSON payload response.
            Defaults to False.

    Returns:
        resp: A JSON formatted response from a HTTP response

    Raises:
        HTTPError: An error was returned by the HTTP request
    """
    resp = requests.patch(url, headers=headers, json=json)

    if not resp:
        raise requests.HTTPError(f"{resp.text}\nRequest URL: {url}")

    if return_json:
        return resp.json()

post_request(url, headers={}, json={}, return_json=False)

Send a POST request to an HTTP API endpoint

Parameters:

Name Type Description Default
url str

The URL to send the request to

 required
headers dict

A dictionary of any headers to send with the request. Defaults to an empty dictionary.

 {}
json dict

A dictionary containing JSON payload to send with the request. Defaults to an empty dictionary.

 {}
return_json bool

Return the JSON payload response. Defaults to False.

 False

Returns:

Name Type Description
resp

A JSON formatted response from a HTTP request

Raises:

Type Description
HTTPError

An error was returned by the HTTP request
Source code in all_all_contributors/http_requests.py 
 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
def post_request(url, headers={}, json={}, return_json=False):
    """Send a POST request to an HTTP API endpoint

    Args:
        url (str): The URL to send the request to
        headers (dict, optional): A dictionary of any headers to send with the
            request. Defaults to an empty dictionary.
        json (dict, optional): A dictionary containing JSON payload to send with
            the request. Defaults to an empty dictionary.
        return_json (bool, optional): Return the JSON payload response.
            Defaults to False.

    Returns:
        resp: A JSON formatted response from a HTTP request

    Raises:
        HTTPError: An error was returned by the HTTP request
    """
    resp = requests.post(url, headers=headers, json=json)

    if not resp:
        raise requests.HTTPError(f"{resp.text}\nRequest URL: {url}")

    if return_json:
        return resp.json()

inject

inject_config(all_contributors_rc, contributors)

Replace the 'contributors' field of an all contributors configuration JSON dict with a new list

Returns:

Name Type Description
dict dict[Any]

Updated .all-contributorsrc config in JSON dict format
Source code in all_all_contributors/inject.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def inject_config(all_contributors_rc: dict[Any], contributors: list[Any]) -> dict[Any]:
    """Replace the 'contributors' field of an all contributors configuration
    JSON dict with a new list

    Returns:
        dict: Updated .all-contributorsrc config in JSON dict format
    """
    if "contributors" in all_contributors_rc.keys():
        all_contributors_rc["contributors"] = contributors
    else:
        raise AttributeError("All contributors file is missing 'contributors' field")

    validate_all_contributors_rc(all_contributors_rc)
    return all_contributors_rc

merge

Merge contributors from multiple .all-contributorsrc files into a single list.

merge_contributions(first, second)

Return a sorted list of the contribution types for two contributor entries

Source code in all_all_contributors/merge.py 
45
46
47
48
49
50
51
52
def merge_contributions(first: Contributor, second: Contributor) -> Contributor:
    """Return a sorted list of the contribution types for two contributor entries"""
    return sorted(
        or_set(
            first.get(_contributions),
            second.get(_contributions),
        )
    )

merge_contributors(contributors_list)

Merge multiple lists of contributor dictionaries into a single list.

This function takes a list of contributor dictionaries (typically from different .all-contributorsrc files) and merges them based on unique profile URLs. When multiple entries exist for the same contributor, their contributions are aggregated into a single entry.

Parameters:

Name Type Description Default
contributors_list list[Contributor]

A list of contributor dictionaries. Each contributor dict should have at least 'profile' and 'contributions' keys.

 required

Returns:

Type Description
list[Contributor]

list[dict[str, Any]]: A list of merged contributor dictionaries, where each contributor appears only once with their combined contributions.
Note

The function merges contributors based on merge._unique_key and aggregates contributions types.

Source code in all_all_contributors/merge.py 
11
12
13
14
15
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
def merge_contributors(contributors_list: list[Contributor]) -> list[Contributor]:
    """Merge multiple lists of contributor dictionaries into a single list.

    This function takes a list of contributor dictionaries (typically from
    different .all-contributorsrc files) and merges them based on unique
    profile URLs. When multiple entries exist for the same contributor, their
    contributions are aggregated into a single entry.

    Args:
        contributors_list: A list of contributor dictionaries. Each contributor
            dict should have at least 'profile' and 'contributions' keys.

    Returns:
        list[dict[str, Any]]: A list of merged contributor dictionaries, where
            each contributor appears only once with their combined contributions.

    Note:
        The function merges contributors based on merge._unique_key and
        aggregates contributions types.
    """

    all_contributors = {}

    for contributor in contributors_list:
        if (key := contributor.get(_unique_key)) in all_contributors.keys():
            all_contributors[key][_contributions] = merge_contributions(
                all_contributors[key], contributor
            )
        else:
            all_contributors[key] = contributor.copy()

    return list(all_contributors.values())

or_set(first, second)

Return list of values that appear in first or second

Source code in all_all_contributors/merge.py 
55
56
57
def or_set(first: list[Any], second: list[Any]) -> list[Any]:
    """Return list of values that appear in `first` or `second`"""
    return list(set(first + second))

validate

validate_all_contributors_rc(all_contributors_rc)

Validate an all contributors configuration object against the schema

Raises:

Type Description
ValidationError

if the configuration is not valid
SchemaError

if the schema is not valid
Source code in all_all_contributors/validate.py 
 7
 8
 9
10
11
12
13
14
15
16
def validate_all_contributors_rc(all_contributors_rc: dict[Any]) -> None:
    """
    Validate an all contributors configuration object against the schema

    Raises:
        ValidationError: if the configuration is not valid
        SchemaError: if the schema is not valid
    """
    schema = requests.get("https://json.schemastore.org/all-contributors.json").json()
    jsonschema.validate(all_contributors_rc, schema)

yaml_parser

Class to safely parse yaml to string and vice versa