forked from smithy-lang/smithy
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
ec7cee5
commit 86715bf
Showing
3 changed files
with
179 additions
and
68 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,166 @@ | ||
| # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
| import json | ||
| import os | ||
| import re | ||
| import time | ||
| from http.client import HTTPResponse | ||
| from pathlib import Path | ||
| from typing import Generator, Literal, NotRequired, Required, TypedDict | ||
| from urllib import request | ||
| from urllib.error import HTTPError | ||
|
|
||
| from . import REPO_ROOT | ||
|
|
||
| GITHUB_API_URL = os.environ.get("GITHUB_API_URL", "https://api.github.com") | ||
| GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN") | ||
| NEXT_PAGE = re.compile(r'(?<=<)([\S]*)(?=>; rel="Next")') | ||
|
|
||
|
|
||
| def post_review_comment( | ||
| repository: str, | ||
| pr_number: str, | ||
| comment: str, | ||
| file: Path, | ||
| start_line: int | None = None, | ||
| end_line: int | None = None, | ||
| allow_duplicate: bool = False, | ||
| ) -> None: | ||
| _assert_token() | ||
|
|
||
| commit_sha = os.environ.get("TARGET_SHA") | ||
| if not commit_sha: | ||
| raise ValueError( | ||
| "The TARGET_SHA environment variable must be set to post review comments." | ||
| ) | ||
|
|
||
| path = str(file.relative_to(REPO_ROOT)) | ||
|
|
||
| if not allow_duplicate: | ||
| for existing_comment in get_review_comments(repository, pr_number): | ||
| if existing_comment["path"] == path and existing_comment["body"] == comment: | ||
| print("Review comment already posted, skipping duplicate.") | ||
| return | ||
|
|
||
| request_body: CreateReviewCommentParams = { | ||
| "body": comment, | ||
| "commit_id": commit_sha, | ||
| "path": path, | ||
| } | ||
|
|
||
| if start_line is not None: | ||
| request_body["side"] = "RIGHT" | ||
| if end_line is not None: | ||
| request_body["start_side"] = "RIGHT" | ||
| request_body["start_line"] = start_line | ||
| request_body["line"] = end_line | ||
| else: | ||
| request_body["line"] = start_line | ||
| elif end_line is not None: | ||
| raise ValueError("If end_line is set, start_line must also be set.") | ||
|
|
||
| body = json.dumps( | ||
| request_body, | ||
| indent=2, | ||
| ) | ||
| resolved_url = f"{GITHUB_API_URL}/repos/{repository}/pulls/{pr_number}/comments" | ||
|
|
||
| print(f"Sending review comment request to {resolved_url} with body:\n\n{body}") | ||
| req = request.Request( | ||
| url=resolved_url, | ||
| headers={ | ||
| "Accept": "application/vnd.github+json", | ||
| "Authorization": f"Bearer {GITHUB_TOKEN}", | ||
| "X-GitHub-Api-Version": "2022-11-28", | ||
| }, | ||
| data=body.encode("utf-8"), | ||
| ) | ||
|
|
||
| try: | ||
| response: HTTPResponse = request.urlopen(req) | ||
| except HTTPError as e: | ||
| formatted_body = json.dumps(json.loads(e.read().decode("utf-8")), indent=4) | ||
| raise Exception(f"Failed to post comment:\n\n{formatted_body}") from e | ||
| print(f"Received response: {response}") | ||
|
|
||
|
|
||
| type ReviewSide = Literal["LEFT"] | Literal["RIGHT"] | ||
|
|
||
|
|
||
| # https://docs.github.com/en/rest/pulls/comments?apiVersion=2022-11-28#create-a-review-comment-for-a-pull-request | ||
| class CreateReviewCommentParams(TypedDict): | ||
| body: Required[str] | ||
| commit_id: Required[str] | ||
| path: Required[str] | ||
| start_line: NotRequired[int] | ||
| line: NotRequired[int] | ||
| start_side: NotRequired[ReviewSide] | ||
| side: NotRequired[ReviewSide] | ||
| subject_type: NotRequired[Literal["line"] | Literal["file"]] | ||
|
|
||
|
|
||
| class User(TypedDict): | ||
| login: Required[str] | ||
|
|
||
|
|
||
| class ReviewComment(TypedDict): | ||
| user: Required[User] | ||
| path: Required[str] | ||
| body: str | ||
|
|
||
|
|
||
| # Use a generator so that we don't necessarily have to request everything if the | ||
| # iterator finds what it wants. | ||
| def get_review_comments( | ||
| repository: str, | ||
| pr_number: str, | ||
| ) -> Generator[ReviewComment]: | ||
| _assert_token() | ||
| url = f"{GITHUB_API_URL}/repos/{repository}/pulls/{pr_number}/comments?per_page=1" | ||
| for response in _paginate(url): | ||
| parsed_body = json.loads(response.read()) | ||
| for comment in parsed_body: | ||
| yield comment | ||
|
|
||
|
|
||
| def _paginate(url: str) -> Generator[HTTPResponse]: | ||
| headers = { | ||
| "Accept": "application/vnd.github+json", | ||
| "Authorization": f"Bearer {GITHUB_TOKEN}", | ||
| "X-GitHub-Api-Version": "2022-11-28", | ||
| } | ||
|
|
||
| print(f"Paginating GitHub api: {url}") | ||
| while True: | ||
| req = request.Request(url=url, headers=headers) | ||
|
|
||
| try: | ||
| response: HTTPResponse = request.urlopen(req) | ||
| except HTTPError as e: | ||
| raise GitHubAPIError(e) from e | ||
|
|
||
| print(f"Received response: {response}") | ||
| yield response | ||
|
|
||
| if match := NEXT_PAGE.match(response.headers.get("link", "")): | ||
| url = match.group() | ||
| print("Waiting one minute before requesting the next page.") | ||
| time.sleep(60) | ||
| else: | ||
| print("Done paginating.") | ||
| break | ||
|
|
||
|
|
||
| class GitHubAPIError(Exception): | ||
| def __init__(self, response: HTTPError) -> None: | ||
| formatted_body = json.dumps( | ||
| json.loads(response.read().decode("utf-8")), indent=2 | ||
| ) | ||
| super().__init__(f"GitHub API request failed:\n\n{formatted_body}") | ||
|
|
||
|
|
||
| def _assert_token(): | ||
| if not GITHUB_TOKEN: | ||
| raise ValueError( | ||
| "The GITHUB_TOKEN environment variable must be set to post review comments." | ||
| ) |