@@ -34,6 +34,14 @@ def __init__(self, github_token: str, update_major_version_only: bool = False):
34
34
self .client = Github (login_or_token = github_token )
35
35
self ._update_major_version_only = update_major_version_only
36
36
37
+ def _get_repo (self , repo_name : str ) -> Repository :
38
+ """Get a repository from github client"""
39
+ try :
40
+ return self .client .get_repo (repo_name )
41
+ except UnknownObjectException :
42
+ logging .error (f"Repository { repo_name } not found" )
43
+ raise ValueError (f"Repository { repo_name } not found" )
44
+
37
45
def _fix_version (self , tag_name : str ) -> str :
38
46
if self ._update_major_version_only :
39
47
return tag_name .split ("." )[0 ]
@@ -77,10 +85,13 @@ def get_action_latest_release(self, uses_tag_value: str) -> Optional[str]:
77
85
logging .debug (f"Current version for { action_name } is a SHA: { current_version } , checking whether latest release is newer" )
78
86
if latest_release [1 ] > datetime .now ():
79
87
return latest_release [0 ]
80
- return latest_release if self ._compare_versions (latest_release [0 ], current_version ) > 0 else None
88
+ return latest_release [ 0 ] if self ._compare_versions (latest_release [0 ], current_version ) > 0 else None
81
89
82
90
logging .debug (f"Checking for updates for { action_name } @{ current_version } : Getting repo { action_name } " )
83
- repo : Repository = self .client .get_repo (action_name )
91
+ try :
92
+ repo : Repository = self ._get_repo (action_name )
93
+ except ValueError as e :
94
+ return None
84
95
logging .info (f"Getting latest release for repository: { action_name } " )
85
96
latest_release : GitRelease
86
97
try :
@@ -125,25 +136,27 @@ def get_workflow_action_names(self, repo_name: str, workflow_path: str) -> Set[s
125
136
126
137
def get_repo_actions_latest (self , repo_name : str ) -> Dict [str , List [ActionVersion ]]:
127
138
workflow_paths = self ._get_github_workflow_filenames (repo_name )
128
- res = dict ()
129
- all_actions = set ()
130
- all_actions_no_version = set () # actions without version, e.g., actions/checkout
139
+ res :Dict [str , List [ActionVersion ]] = dict ()
140
+ actions_per_path :Dict [str ,Set [str ]]= dict () # actions without version, e.g., actions/checkout
131
141
for path in workflow_paths :
132
142
res [path ] = list ()
133
143
actions = self .get_workflow_action_names (repo_name , path )
134
- for action in all_actions :
144
+ for action in actions :
145
+ actions_per_path .setdefault (path ,set ()).add (action )
146
+ all_actions_no_version = set ()
147
+ for path , actions in actions_per_path .items ():
148
+ for action in actions :
135
149
if "@" not in action :
136
150
continue
137
- action_name , _ = action .split ("@" )
138
- all_actions_no_version .add (action_name )
139
- all_actions = all_actions .union (actions )
151
+ all_actions_no_version .add (action .split ("@" )[0 ])
140
152
logging .info (f"Found { len (all_actions_no_version )} actions in workflows: { ", " .join (all_actions_no_version )} " )
141
- for action in all_actions :
142
- if "@" not in action :
143
- continue
144
- action_name , curr_version = action .split ("@" )
145
- latest_version = self .get_action_latest_release (action )
146
- res [path ].append (ActionVersion (action_name , curr_version , latest_version ))
153
+ for path , actions in actions_per_path .items ():
154
+ for action in actions :
155
+ if "@" not in action :
156
+ continue
157
+ action_name , curr_version = action .split ("@" )
158
+ latest_version = self .get_action_latest_release (action )
159
+ res [path ].append (ActionVersion (action_name , curr_version , latest_version ))
147
160
return res
148
161
149
162
def get_repo_workflow_names (self , repo_name : str ) -> Dict [str , str ]:
@@ -184,7 +197,7 @@ def _update_workflow_content(self, repo_name: str, workflow_path: str, workflow_
184
197
return
185
198
186
199
# remote
187
- repo : Repository = self .client . get_repo (repo_name )
200
+ repo : Repository = self ._get_repo (repo_name )
188
201
current_content = repo .get_contents (workflow_path )
189
202
res = repo .update_file (
190
203
workflow_path ,
@@ -205,7 +218,7 @@ def _get_github_workflow_filenames(self, repo_name: str) -> Set[str]:
205
218
click .secho (f"{ repo_name } is not a local repo and does not start with owner/repo" , fg = "red" , err = True )
206
219
raise ValueError (f"{ repo_name } is not a local repo and does not start with owner/repo" )
207
220
# Remote
208
- repo : Repository = self .client . get_repo (repo_name )
221
+ repo : Repository = self ._get_repo (repo_name )
209
222
self ._wf_cache [repo_name ] = {wf .path : wf for wf in repo .get_workflows () if wf .path .startswith (".github/" )}
210
223
return set (self ._wf_cache [repo_name ].keys ())
211
224
@@ -228,8 +241,8 @@ def _get_workflow_file_content(self, repo_name: str, workflow_path: str) -> Unio
228
241
f"possible values: { workflow_paths } " ,
229
242
err = True ,
230
243
)
231
- repo : Repository = self .client .get_repo (repo_name )
232
244
try :
245
+ repo : Repository = self ._get_repo (repo_name )
233
246
workflow_content = repo .get_contents (workflow_path )
234
247
except UnknownObjectException :
235
248
raise FileNotFoundError (f"Workflow not found in repository: { repo_name } , path: { workflow_path } " )
0 commit comments