FilesExpand file tree

 main

/

github_utils.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

235 lines (194 loc) · 7.1 KB

 main

/

github_utils.py

Copy path

Top

File metadata and controls

• Code
• Blame

235 lines (194 loc) · 7.1 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

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

"""GitHub Utilities"""

from __future__ import annotations

import json

import os

import warnings

from dataclasses import dataclass

from typing import Any, cast, TYPE_CHECKING

from urllib.error import HTTPError

from urllib.parse import quote

from urllib.request import Request, urlopen

if TYPE_CHECKING:

from collections.abc import Callable

GITHUB_API_URL = "https://api.github.com"

@dataclass

class GitHubComment:

body_text: str

created_at: str

author_login: str

author_url: str | None

author_association: str

editor_login: str | None

database_id: int

url: str

def gh_fetch_url_and_headers(

url: str,

*,

headers: dict[str, str] | None = None,

data: dict[str, Any] | None | str = None,

method: str | None = None,

reader: Callable[[Any], Any] = lambda x: x.read(),

) -> tuple[Any, Any]:

if headers is None:

headers = {}

token = os.environ.get("GITHUB_TOKEN")

if token is not None and url.startswith(f"{GITHUB_API_URL}/"):

headers["Authorization"] = f"token {token}"

data_ = None

if data is not None:

data_ = data.encode() if isinstance(data, str) else json.dumps(data).encode()

try:

with urlopen(Request(url, headers=headers, data=data_, method=method)) as conn:

return conn.headers, reader(conn)

except HTTPError as err:

if (

err.code == 403

and all(

key in err.headers

for key in ["X-RateLimit-Limit", "X-RateLimit-Remaining"]

)

and int(err.headers["X-RateLimit-Remaining"]) == 0

):

print(

f"""{url}

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"]}"""

)

else:

print(f"Error fetching {url} {err}")

raise

def gh_fetch_url(

url: str,

*,

headers: dict[str, str] | None = None,

data: dict[str, Any] | None | str = None,

method: str | None = None,

reader: Callable[[Any], Any] = json.load,

) -> Any:

return gh_fetch_url_and_headers(

url, headers=headers, data=data, reader=reader, method=method

)[1]

def gh_fetch_json(

url: str,

params: dict[str, Any] | None = None,

data: dict[str, Any] | None = None,

method: str | None = 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: dict[str, Any] | None = None,

data: dict[str, Any] | None = 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: dict[str, Any] | None = None,

data: dict[str, Any] | None = 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: dict[str, Any] | None = None,

data: dict[str, Any] | None = None,

) -> dict[str, Any]:

return cast(dict[str, Any], _gh_fetch_json_any(url, params, data))

def gh_graphql(query: str, **kwargs: Any) -> dict[str, Any]:

rc = gh_fetch_url(

"https://api.github.com/graphql", # @lint-ignore

data={"query": query, "variables": kwargs},

reader=json.load,

)

if "errors" in rc:

raise RuntimeError(

f"GraphQL query {query}, args {kwargs} failed: {rc['errors']}"

)

return cast(dict[str, Any], rc)

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"{GITHUB_API_URL}/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"{GITHUB_API_URL}/repos/{org}/{repo}/commits/{sha}/comments",

comment,

dry_run,

)

def gh_close_pr(org: str, repo: str, pr_num: int, dry_run: bool = False) -> None:

url = f"{GITHUB_API_URL}/repos/{org}/{repo}/pulls/{pr_num}"

if dry_run:

print(f"Dry run closing PR {pr_num}")

else:

gh_fetch_url(url, method="PATCH", data={"state": "closed"})

def gh_delete_comment(org: str, repo: str, comment_id: int) -> None:

url = f"{GITHUB_API_URL}/repos/{org}/{repo}/issues/comments/{comment_id}"

gh_fetch_url(url, method="DELETE", reader=lambda x: x.read())

def gh_fetch_merge_base(org: str, repo: str, base: str, head: str) -> str:

merge_base = ""

# Get the merge base using the GitHub REST API. This is the same as using

# git merge-base without the need to have git. The API doc can be found at

# https://docs.github.com/en/rest/commits/commits?apiVersion=2022-11-28#compare-two-commits

try:

json_data = gh_fetch_url(

f"{GITHUB_API_URL}/repos/{org}/{repo}/compare/{base}...{head}",

headers={"Accept": "application/vnd.github.v3+json"},

reader=json.load,

)

if json_data:

merge_base = json_data.get("merge_base_commit", {}).get("sha", "")

else:

warnings.warn(

f"Failed to get merge base for {base}...{head}: Empty response"

)

except Exception as error:

warnings.warn(f"Failed to get merge base for {base}...{head}: {error}")

return merge_base

def gh_update_pr_state(org: str, repo: str, pr_num: int, state: str = "open") -> None:

url = f"{GITHUB_API_URL}/repos/{org}/{repo}/pulls/{pr_num}"

try:

gh_fetch_url(url, method="PATCH", data={"state": state})

except HTTPError as err:

# When trying to open the pull request, error 422 means that the branch

# has been deleted and the API couldn't re-open it

if err.code == 422 and state == "open":

warnings.warn(

f"Failed to open {pr_num} because its head branch has been deleted: {err}"

)

else:

raise

def gh_query_issues_by_labels(

org: str, repo: str, labels: list[str], state: str = "open"

) -> list[dict[str, Any]]:

url = f"{GITHUB_API_URL}/repos/{org}/{repo}/issues"

return gh_fetch_json(

url, method="GET", params={"labels": ",".join(labels), "state": state}

)