FilesExpand file tree

master

/

github_utils.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

131 lines (109 loc) · 3.85 KB

master

/

github_utils.py

Copy path

Top

File metadata and controls

• Code
• Blame

131 lines (109 loc) · 3.85 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

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

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

"""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:

Used: {err.headers['X-RateLimit-Used']}

Limit: {err.headers['X-RateLimit-Limit']}

Remaining: {err.headers['X-RateLimit-Remaining']}

Resets at: {err.headers['x-RateLimit-Reset']}"""

)

raise

def gh_fetch_json(

url: str,

params: Optional[Dict[str, Any]] = None,

data: Optional[Dict[str, Any]] = None,

method: Optional[str] = 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, method=method),

)

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