Skip to content

github_api.py

Get a GitHub App installation access token.

GithubAdapter

Bases: GitAdapter

GitHub API configuration.

Source code in taglyatelle/git_providers/github_api.py 
 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
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
class GithubAdapter(GitAdapter):
    """GitHub API configuration."""

    def __init__(self, owner: str, repo: str):
        """
        Initialize the GitHub adapter.

        Parameters
        ----------
        owner
            Repository owner or organization

        repo
            Repository name
        """
        self.owner = owner
        self.repo = repo
        self.tag: str | None = None

        self.app_id = os.getenv("APP_ID")
        self.installation_id = os.getenv("INSTALLATION_ID")
        with open(str(os.getenv("PRIVATE_KEY_PATH"))) as f:
            self.private_key = f.read()

        self.access_token = self._access_token()

    def _access_token(self, duration: int = 60) -> str:
        """
        Get an access token.

        Parameters
        ----------
        duration
            time in seconds for which the token is valid (600 seconds by default)

        Returns
        -------
        access token
        """
        now = int(time.time())
        payload = {"iat": now, "exp": now + (10 * duration), "iss": self.app_id}

        jwt_token = jwt.encode(payload, self.private_key, algorithm="RS256")
        headers = {
            "Authorization": f"Bearer {jwt_token}",
            "Accept": "application/vnd.github+json",
        }

        url = f"https://api.github.com/app/installations/{self.installation_id}/access_tokens"
        res = requests.post(url, headers=headers)
        access_token = res.json()["token"]

        return access_token

    def _patch_request(self, url: str, body: dict) -> requests.Response:
        """
        Send a PATCH request to the GitHub API.

        Parameters
        ----------
        url
            GitHub API endpoint

        body
            Request body

        Returns
        -------
        Response from the GitHub API
        """
        api_url = f"https://api.github.com/repos/{self.owner}/{self.repo}/{url}"
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/vnd.github+json",
        }
        response = requests.patch(api_url, headers=headers, json=body)
        return response

    def _post_request(self, url: str, body: dict) -> requests.Response:
        """
        Send a POST request to the GitHub API.

        Parameters
        ----------
        url
            GitHub API endpoint

        body
            Request body

        Returns
        -------
        Response from the GitHub API
        """
        api_url = f"https://api.github.com/repos/{self.owner}/{self.repo}/{url}"
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/vnd.github+json",
        }
        response = requests.post(api_url, headers=headers, json=body)
        return response

    def _get_request(self, url: str) -> requests.Response:
        """
        Send a GET request to the GitHub API.

        Parameters
        ----------
        url
            GitHub API endpoint

        Returns
        -------
        Response from the GitHub API
        """
        api_url = f"https://api.github.com/repos/{self.owner}/{self.repo}/{url}"
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/vnd.github+json",
        }
        response = requests.get(api_url, headers=headers)
        return response

    def get_pr_files(self, pr_number: int) -> list[dict[str, str | int]]:
        """
        Get the list of files changed in a pull request.

        Parameters
        ----------
        pr_number
            Pull request number

        Returns
        -------
        List of files changed in the pull request
        """
        response = self._get_request(url=f"pulls/{pr_number}/files")
        return response.json()

    def get_pr_body(self, pr_number: int) -> str:
        """
        Get the body/description of a pull request.

        Parameters
        ----------
        pr_number
            Pull request number

        Returns
        -------
        The PR body text or None if not present
        """
        response = self._get_request(url=f"pulls/{pr_number}")
        data = response.json()
        return str(data["body"])

    def get_pr_details(self, pr_number: int) -> dict:
        """
        Get the full details of a pull request.

        Parameters
        ----------
        pr_number
            Pull request number

        Returns
        -------
        Dictionary containing full PR details including head, base, title, etc.
        """
        response = self._get_request(url=f"pulls/{pr_number}")
        return response.json()

    def create_pr_body(self, pr_number: int, body: str) -> None:
        """
        Fill in the description body of a pull request.

        Parameters
        ----------
        pr_number
            Pull request number

        body
            Pull request body text
        """
        data = {"body": body}
        self._patch_request(url=f"pulls/{pr_number}", body=data)
        logging.info("PR body has been successfully updated.")

    def create_pr_comment(self, pr_number: int, message: str) -> None:
        """
        Create a comment on a pull request.

        Parameters
        ----------
        pr_number
            Pull request number

        message
            Comment message
        """
        data = {"body": message}
        self._post_request(url=f"issues/{pr_number}/comments", body=data)
        logging.info("PR comment has been successfully delivered.")

    def get_current_tag(self) -> None | str:
        """
        Get the current tag of the repository.

        Returns
        -------
        Current version of the repository
        """
        response = self._get_request(url="tags")
        tags = response.json()
        if tags:
            return tags[0]["name"]
        return None

    def create_tag(self, tag: str) -> None:
        """
        Create a tag in the repository.

        Parameters
        ----------
        tag
            version of the repo
        """
        self.tag = tag
        data = {"ref": f"refs/tags/{tag}", "sha": "main"}
        self._post_request(url="git/refs", body=data)
        logging.info(f"New Tag {tag} has been successfully delivered.")

    def create_release(self, body: str) -> None:
        """
        Create a release in the repository.

        Parameters
        ----------
        body
            Release notes
        """
        data = {
            "tag_name": self.tag,
            "name": self.tag,
            "body": body,
            "draft": False,
            "prerelease": False,
        }
        self._post_request(url="releases", body=data)
        logging.info(f"New Release for Tag {self.tag} has been successfully delivered.")

    def create_issue(self, title: str, body: str, labels: list[str] | None = None) -> int:
        """
        Create an issue in the repository.

        Parameters
        ----------
        title
            Issue title

        body
            Issue body text

        labels
            List of label names to apply to the issue

        Returns
        -------
        Created issue number
        """
        data: dict[str, str | list[str]] = {"title": title, "body": body}
        if labels is not None:
            data["labels"] = labels

        response = self._post_request(url="issues", body=data)
        issue_data = response.json()
        logging.info(f"Issue #{issue_data.get('number')} has been successfully created.")

        return issue_data["number"]

    def search_issues(self, query: str, state: str = "open", labels: list[str] | None = None) -> list[dict]:
        """
        Search for issues in the repository.

        Parameters
        ----------
        query
            Search query string to match in issue titles

        state
            Issue state: 'open', 'closed', or 'all'

        labels
            Filter by label names

        Returns
        -------
        List of matching issues
        """
        params = {"state": state}
        if labels:
            params["labels"] = ",".join(labels)

        response = self._get_request(url=f"issues?{self._build_query_string(params)}")
        issues = response.json()

        if query:
            issues = [issue for issue in issues if query.lower() in issue["title"].lower()]

        return issues

    def update_issue(
        self,
        issue_number: int,
        title: str | None = None,
        body: str | None = None,
        state: str | None = None,
    ) -> int:
        """
        Update an existing issue.

        Parameters
        ----------
        issue_number
            Issue number to update

        title
            New title (optional)

        body
            New body content (optional)

        state
            New state: 'open' or 'closed' (optional)

        Returns
        -------
        Created issue number
        """
        data = {}
        if title is not None:
            data["title"] = title
        if body is not None:
            data["body"] = body
        if state is not None:
            data["state"] = state

        response = self._patch_request(url=f"issues/{issue_number}", body=data)
        issue_data = response.json()
        logging.info(f"Issue #{issue_data.get('number')} has been successfully updated.")

        return issue_data["number"]

    def _build_query_string(self, params: dict) -> str:
        """
        Build URL query string from parameters.

        Parameters
        ----------
        params
            Dictionary of query parameters

        Returns
        -------
        URL encoded query string
        """
        return "&".join(f"{key}={value}" for key, value in params.items() if value)

    def get_file_content(self, file_path: str, ref: str = "main") -> str | None:
        """
        Get the content of a file from the repository.

        Parameters
        ----------
        file_path
            Path to the file in the repository

        ref
            Git reference (branch, tag, or commit SHA)

        Returns
        -------
        Decoded file content or None if file not found
        """
        response = self._get_request(url=f"contents/{file_path}?ref={ref}")
        if response.status_code == 200:
            content_data = response.json()
            encoded_content = content_data.get("content", "")
            return base64.b64decode(encoded_content).decode("utf-8")
        return None

    def get_repository_tree(self, ref: str = "main") -> list[str]:
        """
        Get the file tree of the repository.

        Parameters
        ----------
        ref
            Git reference (branch, tag, or commit SHA)

        Returns
        -------
        List of file paths in the repository
        """
        response = self._get_request(url=f"git/trees/{ref}?recursive=1")
        if response.status_code == 200:
            tree_data = response.json()
            files = []
            for item in tree_data.get("tree", []):
                if item.get("type") == "blob":
                    files.append(item.get("path", ""))
            return files
        return []

__init__(owner, repo)

Initialize the GitHub adapter.

Parameters

owner Repository owner or organization

repo Repository name

Source code in taglyatelle/git_providers/github_api.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(self, owner: str, repo: str):
    """
    Initialize the GitHub adapter.

    Parameters
    ----------
    owner
        Repository owner or organization

    repo
        Repository name
    """
    self.owner = owner
    self.repo = repo
    self.tag: str | None = None

    self.app_id = os.getenv("APP_ID")
    self.installation_id = os.getenv("INSTALLATION_ID")
    with open(str(os.getenv("PRIVATE_KEY_PATH"))) as f:
        self.private_key = f.read()

    self.access_token = self._access_token()

create_issue(title, body, labels=None)

Create an issue in the repository.

Parameters

title Issue title

body Issue body text

labels List of label names to apply to the issue

Returns

Created issue number

Source code in taglyatelle/git_providers/github_api.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
def create_issue(self, title: str, body: str, labels: list[str] | None = None) -> int:
    """
    Create an issue in the repository.

    Parameters
    ----------
    title
        Issue title

    body
        Issue body text

    labels
        List of label names to apply to the issue

    Returns
    -------
    Created issue number
    """
    data: dict[str, str | list[str]] = {"title": title, "body": body}
    if labels is not None:
        data["labels"] = labels

    response = self._post_request(url="issues", body=data)
    issue_data = response.json()
    logging.info(f"Issue #{issue_data.get('number')} has been successfully created.")

    return issue_data["number"]

create_pr_body(pr_number, body)

Fill in the description body of a pull request.

Parameters

pr_number Pull request number

body Pull request body text

Source code in taglyatelle/git_providers/github_api.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def create_pr_body(self, pr_number: int, body: str) -> None:
    """
    Fill in the description body of a pull request.

    Parameters
    ----------
    pr_number
        Pull request number

    body
        Pull request body text
    """
    data = {"body": body}
    self._patch_request(url=f"pulls/{pr_number}", body=data)
    logging.info("PR body has been successfully updated.")

create_pr_comment(pr_number, message)

Create a comment on a pull request.

Parameters

pr_number Pull request number

message Comment message

Source code in taglyatelle/git_providers/github_api.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def create_pr_comment(self, pr_number: int, message: str) -> None:
    """
    Create a comment on a pull request.

    Parameters
    ----------
    pr_number
        Pull request number

    message
        Comment message
    """
    data = {"body": message}
    self._post_request(url=f"issues/{pr_number}/comments", body=data)
    logging.info("PR comment has been successfully delivered.")

create_release(body)

Create a release in the repository.

Parameters

body Release notes

Source code in taglyatelle/git_providers/github_api.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def create_release(self, body: str) -> None:
    """
    Create a release in the repository.

    Parameters
    ----------
    body
        Release notes
    """
    data = {
        "tag_name": self.tag,
        "name": self.tag,
        "body": body,
        "draft": False,
        "prerelease": False,
    }
    self._post_request(url="releases", body=data)
    logging.info(f"New Release for Tag {self.tag} has been successfully delivered.")

create_tag(tag)

Create a tag in the repository.

Parameters

tag version of the repo

Source code in taglyatelle/git_providers/github_api.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
def create_tag(self, tag: str) -> None:
    """
    Create a tag in the repository.

    Parameters
    ----------
    tag
        version of the repo
    """
    self.tag = tag
    data = {"ref": f"refs/tags/{tag}", "sha": "main"}
    self._post_request(url="git/refs", body=data)
    logging.info(f"New Tag {tag} has been successfully delivered.")

get_current_tag()

Get the current tag of the repository.

Returns

Current version of the repository

Source code in taglyatelle/git_providers/github_api.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
def get_current_tag(self) -> None | str:
    """
    Get the current tag of the repository.

    Returns
    -------
    Current version of the repository
    """
    response = self._get_request(url="tags")
    tags = response.json()
    if tags:
        return tags[0]["name"]
    return None

get_file_content(file_path, ref='main')

Get the content of a file from the repository.

Parameters

file_path Path to the file in the repository

ref Git reference (branch, tag, or commit SHA)

Returns

Decoded file content or None if file not found

Source code in taglyatelle/git_providers/github_api.py 
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
def get_file_content(self, file_path: str, ref: str = "main") -> str | None:
    """
    Get the content of a file from the repository.

    Parameters
    ----------
    file_path
        Path to the file in the repository

    ref
        Git reference (branch, tag, or commit SHA)

    Returns
    -------
    Decoded file content or None if file not found
    """
    response = self._get_request(url=f"contents/{file_path}?ref={ref}")
    if response.status_code == 200:
        content_data = response.json()
        encoded_content = content_data.get("content", "")
        return base64.b64decode(encoded_content).decode("utf-8")
    return None

get_pr_body(pr_number)

Get the body/description of a pull request.

Parameters

pr_number Pull request number

Returns

The PR body text or None if not present

Source code in taglyatelle/git_providers/github_api.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def get_pr_body(self, pr_number: int) -> str:
    """
    Get the body/description of a pull request.

    Parameters
    ----------
    pr_number
        Pull request number

    Returns
    -------
    The PR body text or None if not present
    """
    response = self._get_request(url=f"pulls/{pr_number}")
    data = response.json()
    return str(data["body"])

get_pr_details(pr_number)

Get the full details of a pull request.

Parameters

pr_number Pull request number

Returns

Dictionary containing full PR details including head, base, title, etc.

Source code in taglyatelle/git_providers/github_api.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def get_pr_details(self, pr_number: int) -> dict:
    """
    Get the full details of a pull request.

    Parameters
    ----------
    pr_number
        Pull request number

    Returns
    -------
    Dictionary containing full PR details including head, base, title, etc.
    """
    response = self._get_request(url=f"pulls/{pr_number}")
    return response.json()

get_pr_files(pr_number)

Get the list of files changed in a pull request.

Parameters

pr_number Pull request number

Returns

List of files changed in the pull request

Source code in taglyatelle/git_providers/github_api.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def get_pr_files(self, pr_number: int) -> list[dict[str, str | int]]:
    """
    Get the list of files changed in a pull request.

    Parameters
    ----------
    pr_number
        Pull request number

    Returns
    -------
    List of files changed in the pull request
    """
    response = self._get_request(url=f"pulls/{pr_number}/files")
    return response.json()

get_repository_tree(ref='main')

Get the file tree of the repository.

Parameters

ref Git reference (branch, tag, or commit SHA)

Returns

List of file paths in the repository

Source code in taglyatelle/git_providers/github_api.py 
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
def get_repository_tree(self, ref: str = "main") -> list[str]:
    """
    Get the file tree of the repository.

    Parameters
    ----------
    ref
        Git reference (branch, tag, or commit SHA)

    Returns
    -------
    List of file paths in the repository
    """
    response = self._get_request(url=f"git/trees/{ref}?recursive=1")
    if response.status_code == 200:
        tree_data = response.json()
        files = []
        for item in tree_data.get("tree", []):
            if item.get("type") == "blob":
                files.append(item.get("path", ""))
        return files
    return []

search_issues(query, state='open', labels=None)

Search for issues in the repository.

Parameters

query Search query string to match in issue titles

state Issue state: 'open', 'closed', or 'all'

labels Filter by label names

Returns

List of matching issues

Source code in taglyatelle/git_providers/github_api.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def search_issues(self, query: str, state: str = "open", labels: list[str] | None = None) -> list[dict]:
    """
    Search for issues in the repository.

    Parameters
    ----------
    query
        Search query string to match in issue titles

    state
        Issue state: 'open', 'closed', or 'all'

    labels
        Filter by label names

    Returns
    -------
    List of matching issues
    """
    params = {"state": state}
    if labels:
        params["labels"] = ",".join(labels)

    response = self._get_request(url=f"issues?{self._build_query_string(params)}")
    issues = response.json()

    if query:
        issues = [issue for issue in issues if query.lower() in issue["title"].lower()]

    return issues

update_issue(issue_number, title=None, body=None, state=None)

Update an existing issue.

Parameters

issue_number Issue number to update

title New title (optional)

body New body content (optional)

state New state: 'open' or 'closed' (optional)

Returns

Created issue number

Source code in taglyatelle/git_providers/github_api.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def update_issue(
    self,
    issue_number: int,
    title: str | None = None,
    body: str | None = None,
    state: str | None = None,
) -> int:
    """
    Update an existing issue.

    Parameters
    ----------
    issue_number
        Issue number to update

    title
        New title (optional)

    body
        New body content (optional)

    state
        New state: 'open' or 'closed' (optional)

    Returns
    -------
    Created issue number
    """
    data = {}
    if title is not None:
        data["title"] = title
    if body is not None:
        data["body"] = body
    if state is not None:
        data["state"] = state

    response = self._patch_request(url=f"issues/{issue_number}", body=data)
    issue_data = response.json()
    logging.info(f"Issue #{issue_data.get('number')} has been successfully updated.")

    return issue_data["number"]