mirror of
https://github.com/zebrajr/pytorch.git
synced 2025-12-07 00:21:07 +01:00
2c76838d7f
Fixes #88098 This is a mirror of the same PR (https://github.com/Goldspear/pytorch/pull/2) that has been reviewed in my fork (due to it's a stacked PR). ====================== ## Context This is the 2nd of the 3 PRs to address issue-88098. ## What Changed 1. Extract comment related utils from trymerge.py to github_utils.py 2. Extract label related utils from trymerge.py and check_labels.py to label_utils.py ## Tests * pytorch-dummy repo [trymerge run ](https://github.com/Goldspear/pytorch-dummy/actions/runs/4118944174)merged the test PR [OK](https://github.com/Goldspear/pytorch-dummy/pull/2). ## Note to Reviewers Due to higher degree of complexity involved to extract GitHubPR class, it's worth having a separate issue to handle that part of refactoring. This issue only focusing on refactoring where necessary to ship the functional diff. * 1st PR: https://github.com/pytorch/pytorch/pull/94179 * 2nd PR: this one * 3rd PR: https://github.com/Goldspear/pytorch/pull/3 Pull Request resolved: https://github.com/pytorch/pytorch/pull/94597 Approved by: https://github.com/ZainRizvi
100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
"""GitHub Utilities"""
|
|
|
|
import json
|
|
import os
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Callable, cast, Dict, List, Optional
|
|
from urllib.error import HTTPError
|
|
from urllib.parse import quote
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
@dataclass
|
|
class GitHubComment:
|
|
body_text: str
|
|
created_at: str
|
|
author_login: str
|
|
author_association: str
|
|
editor_login: Optional[str]
|
|
database_id: int
|
|
|
|
|
|
def gh_fetch_url(
|
|
url: str, *,
|
|
headers: Optional[Dict[str, str]] = None,
|
|
data: Optional[Dict[str, Any]] = None,
|
|
method: Optional[str] = None,
|
|
reader: Callable[[Any], Any] = lambda x: x.read()
|
|
) -> Any:
|
|
if headers is None:
|
|
headers = {}
|
|
token = os.environ.get("GITHUB_TOKEN")
|
|
if token is not None and url.startswith('https://api.github.com/'):
|
|
headers['Authorization'] = f'token {token}'
|
|
data_ = json.dumps(data).encode() if data is not None else None
|
|
try:
|
|
with urlopen(Request(url, headers=headers, data=data_, method=method)) as conn:
|
|
return reader(conn)
|
|
except HTTPError as err:
|
|
if err.code == 403 and all(key in err.headers for key in ['X-RateLimit-Limit', 'X-RateLimit-Used']):
|
|
print(f"Rate limit exceeded: {err.headers['X-RateLimit-Used']}/{err.headers['X-RateLimit-Limit']}")
|
|
raise
|
|
|
|
|
|
def gh_fetch_json(
|
|
url: str,
|
|
params: Optional[Dict[str, Any]] = None,
|
|
data: Optional[Dict[str, Any]] = None
|
|
) -> List[Dict[str, Any]]:
|
|
headers = {'Accept': 'application/vnd.github.v3+json'}
|
|
if params is not None and len(params) > 0:
|
|
url += '?' + '&'.join(f"{name}={quote(str(val))}" for name, val in params.items())
|
|
return cast(List[Dict[str, Any]], gh_fetch_url(url, headers=headers, data=data, reader=json.load))
|
|
|
|
def _gh_fetch_json_any(
|
|
url: str,
|
|
params: Optional[Dict[str, Any]] = None,
|
|
data: Optional[Dict[str, Any]] = None
|
|
) -> Any:
|
|
headers = {'Accept': 'application/vnd.github.v3+json'}
|
|
if params is not None and len(params) > 0:
|
|
url += '?' + '&'.join(f"{name}={quote(str(val))}" for name, val in params.items())
|
|
return gh_fetch_url(url, headers=headers, data=data, reader=json.load)
|
|
|
|
|
|
def gh_fetch_json_list(
|
|
url: str,
|
|
params: Optional[Dict[str, Any]] = None,
|
|
data: Optional[Dict[str, Any]] = None
|
|
) -> List[Dict[str, Any]]:
|
|
return cast(List[Dict[str, Any]], _gh_fetch_json_any(url, params, data))
|
|
|
|
|
|
def gh_fetch_json_dict(
|
|
url: str,
|
|
params: Optional[Dict[str, Any]] = None,
|
|
data: Optional[Dict[str, Any]] = None
|
|
) -> Dict[str, Any] :
|
|
return cast(Dict[str, Any], _gh_fetch_json_any(url, params, data))
|
|
|
|
|
|
def _gh_post_comment(url: str, comment: str, dry_run: bool = False) -> List[Dict[str, Any]]:
|
|
if dry_run:
|
|
print(comment)
|
|
return []
|
|
return gh_fetch_json_list(url, data={"body": comment})
|
|
|
|
|
|
def gh_post_pr_comment(org: str, repo: str, pr_num: int, comment: str, dry_run: bool = False) -> List[Dict[str, Any]]:
|
|
return _gh_post_comment(f'https://api.github.com/repos/{org}/{repo}/issues/{pr_num}/comments', comment, dry_run)
|
|
|
|
|
|
def gh_post_commit_comment(org: str, repo: str, sha: str, comment: str, dry_run: bool = False) -> List[Dict[str, Any]]:
|
|
return _gh_post_comment(f'https://api.github.com/repos/{org}/{repo}/commits/{sha}/comments', comment, dry_run)
|
|
|
|
|
|
def gh_delete_comment(org: str, repo: str, comment_id: int) -> None:
|
|
url = f"https://api.github.com/repos/{org}/{repo}/issues/comments/{comment_id}"
|
|
gh_fetch_url(url, method="DELETE")
|