1
1
import os
2
+
2
3
import requests
3
4
4
- def get_github_prs (token , owner , repo , label = "" , state = "all" ):
5
+
6
+ def get_github_prs (token : str , owner : str , repo : str , label : str = "" , state : str = "all" ) -> list [dict ]:
5
7
"""
6
8
Fetches pull requests from a GitHub repository that match a given milestone and label.
7
9
@@ -10,7 +12,7 @@ def get_github_prs(token, owner, repo, label = "", state ="all"):
10
12
owner (str): The owner of the repository.
11
13
repo (str): The name of the repository.
12
14
label (str): The label name.
13
- state (str): State of PR, e.g. open
15
+ state (str): State of PR, e.g. open, close, all
14
16
15
17
Returns:
16
18
list: A list of dictionaries, where each dictionary represents a pull request.
@@ -20,11 +22,11 @@ def get_github_prs(token, owner, repo, label = "", state ="all"):
20
22
"Authorization" : f"token { token } " ,
21
23
"Accept" : "application/vnd.github.v3+json" ,
22
24
}
23
-
25
+
24
26
milestone_id = None
25
27
milestone_url = f"https://api.github.com/repos/{ owner } /{ repo } /milestones"
26
28
params = {"state" : open }
27
-
29
+
28
30
try :
29
31
response = requests .get (milestone_url , headers = headers , params = params )
30
32
response .raise_for_status ()
@@ -37,9 +39,9 @@ def get_github_prs(token, owner, repo, label = "", state ="all"):
37
39
for ms in milestones :
38
40
if ms ["title" ] != "Future" :
39
41
milestone_id = ms ["number" ]
40
- print (f"Gathering PRs with milestone { ms ['title' ]} ..." )
42
+ print (f"Gathering PRs with milestone { ms ['title' ]} ..." )
41
43
break
42
-
44
+
43
45
if not milestone_id :
44
46
print (f"No suitable milestone found in repository '{ owner } /{ repo } '." )
45
47
exit (1 )
@@ -63,42 +65,70 @@ def get_github_prs(token, owner, repo, label = "", state ="all"):
63
65
try :
64
66
params ["page" ] = page
65
67
response = requests .get (prs_url , headers = headers , params = params )
66
- response .raise_for_status () # Raise an exception for HTTP errors
68
+ response .raise_for_status () # Raise an exception for HTTP errors
67
69
prs = response .json ()
68
-
70
+
69
71
if not prs :
70
- break # No more PRs to fetch
71
-
72
+ break # No more PRs to fetch
73
+
72
74
# Check for pr key since we are using issues endpoint instead.
73
75
all_prs .extend ([item for item in prs if "pull_request" in item ])
74
76
page += 1
75
77
76
78
except requests .exceptions .RequestException as e :
77
79
print (f"Error fetching pull requests: { e } " )
78
80
exit (1 )
79
-
81
+
80
82
return all_prs
81
83
82
- def get_prs (pull_request_items , label = "" , state = "all" ):
84
+
85
+ def get_prs (pull_request_items : list [dict ], label : str = "" , state : str = "all" ) -> list [dict ]:
86
+ """
87
+ Returns a list of pull requests after applying the label and state filters.
88
+
89
+ Args:
90
+ pull_request_items (str): List of PR items.
91
+ label (str): The label name.
92
+ state (str): State of PR, e.g. open, close, all
93
+
94
+ Returns:
95
+ list: A list of dictionaries, where each dictionary represents a pull request.
96
+ Returns an empty list if no PRs are found.
97
+ """
83
98
pr_list = []
84
99
count = 0
85
100
for pr in pull_request_items :
86
101
if pr ["state" ] == state and [item for item in pr ["labels" ] if item ["name" ] == label ]:
87
102
pr_list .append (pr )
88
103
count += 1
89
-
104
+
90
105
print (f"Found { count } PRs with { label if label else "no" } label and state as { state } " )
91
106
92
107
return pr_list
93
108
94
- def get_pr_descriptions (pull_request_items ):
109
+
110
+ def get_pr_descriptions (pull_request_items : list [dict ]) -> str :
111
+ """
112
+ Returns the concatenated string of pr title and number in the format of
113
+ '- PR title 1 #3651
114
+ - PR title 2 #3652
115
+ - PR title 3 #3653
116
+ '
117
+
118
+ Args:
119
+ pull_request_items (list[dict]): List of PR items.
120
+
121
+ Returns:
122
+ str: a string of PR titles and numbers
123
+ """
95
124
description_content = ""
96
125
for pr in pull_request_items :
97
- description_content += f"- { pr ['title' ]} #{ pr ['number' ]} \n "
126
+ description_content += f"- { pr ['title' ]} #{ pr ['number' ]} \n "
98
127
99
128
return description_content
100
129
101
- def update_pull_request_description (token , owner , repo , pr_number , new_description ):
130
+
131
+ def update_pull_request_description (token : str , owner : str , repo : str , pr_number : int , new_description : str ) -> None :
102
132
"""
103
133
Updates the description (body) of a GitHub Pull Request.
104
134
@@ -116,14 +146,12 @@ def update_pull_request_description(token, owner, repo, pr_number, new_descripti
116
146
headers = {
117
147
"Authorization" : f"token { token } " ,
118
148
"Accept" : "application/vnd.github.v3+json" ,
119
- "Content-Type" : "application/json"
149
+ "Content-Type" : "application/json" ,
120
150
}
121
151
122
152
url = f"https://api.github.com/repos/{ owner } /{ repo } /pulls/{ pr_number } "
123
153
124
- payload = {
125
- "body" : new_description
126
- }
154
+ payload = {"body" : new_description }
127
155
128
156
print (f"Attempting to update PR #{ pr_number } in { owner } /{ repo } ..." )
129
157
print (f"URL: { url } " )
@@ -132,21 +160,19 @@ def update_pull_request_description(token, owner, repo, pr_number, new_descripti
132
160
response = requests .patch (url , headers = headers , json = payload )
133
161
response .raise_for_status ()
134
162
135
- updated_pr_data = response .json ()
136
163
print (f"Successfully updated PR #{ pr_number } ." )
137
- return updated_pr_data
138
164
139
165
except requests .exceptions .RequestException as e :
140
166
print (f"Error updating pull request #{ pr_number } : { e } " )
141
167
if response is not None :
142
168
print (f"Response status code: { response .status_code } " )
143
169
print (f"Response text: { response .text } " )
144
- return None
170
+ exit ( 1 )
145
171
146
172
147
173
if __name__ == "__main__" :
148
- github_token = os .environ .get ("GITHUB_TOKEN " )
149
-
174
+ github_token = os .environ .get ("PR_GET_TOKEN " )
175
+
150
176
if not github_token :
151
177
print ("Error: GITHUB_TOKEN environment variable not set." )
152
178
exit (1 )
@@ -158,28 +184,20 @@ def update_pull_request_description(token, owner, repo, pr_number, new_descripti
158
184
159
185
print (f"Fetching PRs for { repository_owner } /{ repository_name } with label '{ target_label } '..." )
160
186
161
- pull_requests = get_github_prs (
162
- github_token ,
163
- repository_owner ,
164
- repository_name
165
- )
187
+ pull_requests = get_github_prs (github_token , repository_owner , repository_name )
166
188
167
189
if not pull_requests :
168
190
print ("No matching pull requests found" )
169
191
exit (1 )
170
192
171
193
print (f"\n Found total of { len (pull_requests )} pull requests" )
172
194
173
- release_pr = get_prs (
174
- pull_requests ,
175
- "release" ,
176
- "open"
177
- )
195
+ release_pr = get_prs (pull_requests , "release" , "open" )
178
196
179
197
if len (release_pr ) != 1 :
180
198
print (f"Unable to find the exact release PR. Returned result: { release_pr } " )
181
199
exit (1 )
182
-
200
+
183
201
print (f"Found release PR: { release_pr [0 ]['title' ]} " )
184
202
185
203
enhancement_prs = get_prs (pull_requests , "enhancement" , "closed" )
@@ -189,6 +207,8 @@ def update_pull_request_description(token, owner, repo, pr_number, new_descripti
189
207
description_content += f"## Features\n { get_pr_descriptions (enhancement_prs )} " if enhancement_prs else ""
190
208
description_content += f"## Bug fixes\n { get_pr_descriptions (bug_fix_prs )} " if bug_fix_prs else ""
191
209
192
- update_pull_request_description (github_token , repository_owner , repository_name , release_pr [0 ]["number" ], description_content )
210
+ update_pull_request_description (
211
+ github_token , repository_owner , repository_name , release_pr [0 ]["number" ], description_content
212
+ )
193
213
194
- print (f"PR content updated to:\n { description_content } " )
214
+ print (f"PR content updated to:\n { description_content } " )
0 commit comments