22import logging
33import os
44from collections import namedtuple
5- from typing import Optional , List , Set , Dict , Union
5+ from typing import Optional , List , Set , Dict , Union , Any
66
77import click
88import yaml
1717FLAG_COMPARE_EXACT_VERSION = False
1818
1919
20+ def compare_versions (v1 : str , v2 : str ) -> int :
21+ """Compare two versions, return 1 if v1 > v2, 0 if v1 == v2, -1 if v1 < v2
22+ """
23+ if v1 .startswith ('v' ):
24+ v1 = v1 [1 :]
25+ if v2 .startswith ('v' ):
26+ v2 = v2 [1 :]
27+ v1 = v1 .split ('.' )
28+ v2 = v2 .split ('.' )
29+ compare_count = max (len (v1 ), len (v2 )) if FLAG_COMPARE_EXACT_VERSION else 1
30+ for i in range (compare_count ):
31+ v1_i = int (v1 [i ]) if i < len (v1 ) else 0
32+ v2_i = int (v2 [i ]) if i < len (v2 ) else 0
33+ if v1_i > v2_i :
34+ return 1
35+ if v1_i < v2_i :
36+ return - 1
37+ return 0
38+
39+
2040class GithubActionsTools (object ):
21- workflows : dict [str , dict [str , Workflow ]] = dict () # repo_name -> [path -> workflow]
41+ _wf_cache : dict [str , dict [str , Union [ Workflow , Any ]]] = dict () # repo_name -> [path -> workflow/yaml ]
2242 actions_latest_release : dict [str , str ] = dict () # action_name@current_release -> latest_release_tag
2343
2444 def __init__ (self , github_token : str ):
2545 self .client = Github (login_or_token = github_token )
2646
27- @staticmethod
28- def compare_versions (v1 : str , v2 : str ) -> int :
29- """Compare two versions, return 1 if v1 > v2, 0 if v1 == v2, -1 if v1 < v2
30- """
31- if v1 .startswith ('v' ):
32- v1 = v1 [1 :]
33- if v2 .startswith ('v' ):
34- v2 = v2 [1 :]
35- v1 = v1 .split ('.' )
36- v2 = v2 .split ('.' )
37- compare_count = max (len (v1 ), len (v2 )) if FLAG_COMPARE_EXACT_VERSION else 1
38- for i in range (compare_count ):
39- v1_i = int (v1 [i ]) if i < len (v1 ) else 0
40- v2_i = int (v2 [i ]) if i < len (v2 ) else 0
41- if v1_i > v2_i :
42- return 1
43- if v1_i < v2_i :
44- return - 1
45- return 0
46-
4747 @staticmethod
4848 def is_local_repo (repo_name : str ) -> bool :
4949 return os .path .exists (repo_name ) and os .path .exists (os .path .join (repo_name , '.git' ))
@@ -56,44 +56,8 @@ def list_full_paths(path: str) -> set[str]:
5656 for file in os .listdir (path )
5757 if file .endswith (('.yml' , '.yaml' ))}
5858
59- def get_github_workflows (self , repo_name : str ) -> Set [str ]:
60- if repo_name in self .workflows :
61- return set (self .workflows [repo_name ].keys ())
62- # local
63- if self .is_local_repo (repo_name ):
64- return self .list_full_paths (os .path .join (repo_name , '.github' , 'workflows' ))
65- if repo_name .startswith ('.' ):
66- click .secho (f'{ repo_name } is not a local repo and does not start with owner/repo' , fg = 'red' , err = True )
67- exit (1 )
68- # Remote
69- repo = self .client .get_repo (repo_name )
70- self .workflows [repo_name ] = {
71- wf .path : wf
72- for wf in repo .get_workflows ()
73- if wf .path .startswith ('.github/' )}
74- return set (self .workflows [repo_name ].keys ())
75-
76- def _get_workflow_content (self , repo_name : str , workflow_path : str ) -> Union [str , bytes ]:
77- workflow_paths = self .get_github_workflows (repo_name )
78-
79- if self .is_local_repo (repo_name ):
80- if not os .path .exists (workflow_path ):
81- click .echo (
82- f'f{ workflow_path } not found in workflows for repository { repo_name } , '
83- f'possible values: { workflow_paths } ' , err = True )
84- with open (workflow_path ) as f :
85- return f .read ()
86-
87- if workflow_path not in workflow_paths :
88- click .echo (
89- f'f{ workflow_path } not found in workflows for repository { repo_name } , '
90- f'possible values: { workflow_paths } ' , err = True )
91- repo = self .client .get_repo (repo_name )
92- workflow_content = repo .get_contents (workflow_path )
93- return workflow_content .decoded_content
94-
9559 def get_workflow_actions (self , repo_name : str , workflow_path : str ) -> Set [str ]:
96- workflow_content = self ._get_workflow_content (repo_name , workflow_path )
60+ workflow_content = self ._get_workflow_file_content (repo_name , workflow_path )
9761 workflow = yaml .load (workflow_content , Loader = yaml .CLoader )
9862 res = set ()
9963 for job in workflow .get ('jobs' , dict ()).values ():
@@ -103,21 +67,20 @@ def get_workflow_actions(self, repo_name: str, workflow_path: str) -> Set[str]:
10367 return res
10468
10569 def check_for_updates (self , action_name : str ) -> Optional [str ]:
106- """Check whether an action has update, and return the latest version if it does
107- syntax for uses:
70+ """Check whether an action has an update, and return the latest version if it does syntax for uses:
10871 https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#jobsjob_iduses
10972 """
11073 if '@' not in action_name :
11174 return None
11275 repo_name , current_version = action_name .split ('@' )
11376 repo = self .client .get_repo (repo_name )
11477 latest_release = repo .get_latest_release ()
115- if GithubActionsTools . compare_versions (latest_release .tag_name , current_version ):
78+ if compare_versions (latest_release .tag_name , current_version ):
11679 return latest_release .tag_name
11780 return None
11881
11982 def get_repo_actions_latest (self , repo_name : str ) -> Dict [str , List [ActionVersion ]]:
120- workflow_paths = self .get_github_workflows (repo_name )
83+ workflow_paths = self ._get_github_workflow_filenames (repo_name )
12184 res = dict ()
12285 for path in workflow_paths :
12386 res [path ] = list ()
@@ -134,12 +97,21 @@ def get_repo_actions_latest(self, repo_name: str) -> Dict[str, List[ActionVersio
13497 res [path ].append (ActionVersion (action_name , curr_version , latest ))
13598 return res
13699
100+ def get_repo_workflow_names (self , repo_name : str ) -> Dict [str , str ]:
101+ workflow_paths = self ._get_github_workflow_filenames (repo_name )
102+ res = dict ()
103+ for path in workflow_paths :
104+ content = self ._get_workflow_file_content (repo_name , path )
105+ yaml_content = yaml .load (content , Loader = yaml .CLoader )
106+ res [path ] = yaml_content .get ('name' , path )
107+ return res
108+
137109 def update_actions (
138110 self , repo_name : str , workflow_path : str ,
139111 updates : List [ActionVersion ],
140112 commit_msg : str ,
141113 ) -> None :
142- workflow_content = self ._get_workflow_content (repo_name , workflow_path )
114+ workflow_content = self ._get_workflow_file_content (repo_name , workflow_path )
143115 if isinstance (workflow_content , bytes ):
144116 workflow_content = workflow_content .decode ()
145117 for update in updates :
@@ -170,6 +142,42 @@ def _update_workflow_content(
170142 click .secho (f'Committed changes to workflow in { repo_name } :{ workflow_path } ' , fg = 'cyan' )
171143 return res
172144
145+ def _get_github_workflow_filenames (self , repo_name : str ) -> Set [str ]:
146+ if repo_name in self ._wf_cache :
147+ return set (self ._wf_cache [repo_name ].keys ())
148+ # local
149+ if self .is_local_repo (repo_name ):
150+ return self .list_full_paths (os .path .join (repo_name , '.github' , 'workflows' ))
151+ if repo_name .startswith ('.' ):
152+ click .secho (f'{ repo_name } is not a local repo and does not start with owner/repo' , fg = 'red' , err = True )
153+ raise ValueError (f'{ repo_name } is not a local repo and does not start with owner/repo' )
154+ # Remote
155+ repo = self .client .get_repo (repo_name )
156+ self ._wf_cache [repo_name ] = {
157+ wf .path : wf
158+ for wf in repo .get_workflows ()
159+ if wf .path .startswith ('.github/' )}
160+ return set (self ._wf_cache [repo_name ].keys ())
161+
162+ def _get_workflow_file_content (self , repo_name : str , workflow_path : str ) -> Union [str , bytes ]:
163+ workflow_paths = self ._get_github_workflow_filenames (repo_name )
164+
165+ if self .is_local_repo (repo_name ):
166+ if not os .path .exists (workflow_path ):
167+ click .echo (
168+ f'f{ workflow_path } not found in workflows for repository { repo_name } , '
169+ f'possible values: { workflow_paths } ' , err = True )
170+ with open (workflow_path ) as f :
171+ return f .read ()
172+
173+ if workflow_path not in workflow_paths :
174+ click .echo (
175+ f'f{ workflow_path } not found in workflows for repository { repo_name } , '
176+ f'possible values: { workflow_paths } ' , err = True )
177+ repo = self .client .get_repo (repo_name )
178+ workflow_content = repo .get_contents (workflow_path )
179+ return workflow_content .decoded_content
180+
173181
174182GITHUB_ACTION_NOT_PROVIDED_MSG = """GitHub connection token not provided.
175183You might not be able to make the changes to remote repositories.
@@ -203,18 +211,19 @@ def cli(ctx, repo: str, github_token: Optional[str], compare_exact_versions: boo
203211@cli .command (help = 'Show actions required updates in repository workflows' )
204212@click .option (
205213 '-u' , '--update' , is_flag = True , default = False ,
206- help = 'Update actions in workflows (For remote repos: make changes and commit, for local repos: update files' ,)
214+ help = 'Update actions in workflows (For remote repos: make changes and commit, for local repos: update files' , )
207215@click .option (
208216 '-commit-msg' ,
209217 default = 'chore(ci):update actions' , type = str , show_default = True ,
210218 help = 'Commit msg, only relevant when remote repo' )
211219@click .pass_context
212220def update_actions (ctx , update : bool , commit_msg : str ):
213221 gh , repo = ctx .obj ['gh' ], ctx .obj ['repo' ]
222+ workflow_names = (gh .get_repo_workflow_names (repo ))
214223 workflow_action_versions = gh .get_repo_actions_latest (repo )
215- for workflow in workflow_action_versions :
216- click .secho (f'{ workflow } :' , fg = 'blue ' )
217- for action in workflow_action_versions [workflow ]:
224+ for workflow_path , workflow_name in workflow_names . items () :
225+ click .secho (f'{ workflow_path } ( { click . style ( workflow_name , fg = "bright_cyan" ) } ) :' , fg = 'bright_blue ' )
226+ for action in workflow_action_versions [workflow_path ]:
218227 s = f'\t { action .name :30} { action .current :>5} '
219228 if action .latest :
220229 old_version = action .current .split ('.' )
@@ -240,9 +249,9 @@ def list_actions(ctx, workflow: str):
240249@cli .command (help = 'List workflows in repository' )
241250@click .pass_context
242251def list_workflows (ctx ):
243- workflow_paths = ctx .obj ['gh' ].get_github_workflows (ctx .obj ['repo' ])
244- for path in workflow_paths :
245- click .echo (f'{ path } ' )
252+ workflow_paths = ( ctx .obj ['gh' ].get_repo_workflow_names (ctx .obj ['repo' ]) )
253+ for path , name in workflow_paths . items () :
254+ click .echo (f'{ path } - { name } ' )
246255
247256
248257if __name__ == '__main__' :
0 commit comments