PyXMake Developer Guide 1.0
PyXMake
Loading...
Searching...
No Matches
__gitlab.py
1# -*- coding: utf-8 -*-
2# %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
3# % GitLab wrapper module - Classes and functions %
4# %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
5"""
6GitLab release configuration and management assistance wrapper.
7
8@note:
9Created on 16.08.2022
10
11@version: 1.0
12----------------------------------------------------------------------------------------------
13@requires:
14 -
15
16@change:
17 -
18
19@author: garb_ma [DLR-FA,STM Braunschweig]
20----------------------------------------------------------------------------------------------
21"""
22import os, sys
23import copy
24import requests
25import posixpath
26import warnings
27import argparse
28
29try: # pragma: no cover
30 from packaging.version import Version as StrictVersion
31 LooseVersion = StrictVersion
32except ImportError: from distutils.version import StrictVersion, LooseVersion
33
34from packaging import version
35from datetime import datetime, timedelta
36
37try: from urllib.parse import urlparse #@UnusedImport
38except: from urlparse import urlparse #@UnresolvedImport @Reimport
39
40def check():
41 """
42 GitLab CI methods are only available in CI environment.
43 """
44 if not os.getenv("GITLAB_CI",""): warnings.warn("Method executed outside Gitlab CI. It will have no effect.", RuntimeWarning) #@UndefinedVariable
45 # Return a boolean value
46 return os.getenv("GITLAB_CI","") != ""
47
48def housekeeping(**kwargs):
49 """
50 Entry point for housekeeping job.
51 """
52 # We are outside GitLab CI. Do nothing.
53 if not check(): return 0
54 # Predefined variables
55 outdated = []; headers = {}
56 # Fetch GitLab CI URL
57 url = "%s/projects/%s/pipelines" % (os.getenv("CI_API_V4_URL"), os.getenv("CI_PROJECT_ID"), )
58 # Update request header
59 headers.update({"PRIVATE-TOKEN":os.getenv("GIT_PASSWORD")})
60 # Get time offset from now. Defaults to 14 days
61 d = datetime.today() - timedelta(days=int(os.getenv("since",7)))
62 # Fetch all outdated IDs. Fail gracefully
63 try: outdated = [ x["id"] for x in requests.get(url, headers=headers, params={"updated_before":d}).json() ]
64 except: pass
65 # Loop over all pipelines
66 for pipeline in outdated:
67 # Attempt to delete the pipeline
68 try: requests.delete(posixpath.join(url,str(pipeline)), headers=headers)
69 except: pass
70 return 0
71
72def release(**kwargs):
73 """
74 Entry point for release job.
75 """
76 # We are outside GitLab CI. Do nothing.
77 if not check(): return 0
78 # Collect all user input
79 header = {"PRIVATE-TOKEN": kwargs.get("token",os.getenv("GIT_PASSWORD",None))}
80 url = kwargs.get("base_url",os.getenv("CI_API_V4_URL",None))
81 project = kwargs.get("project_url",os.getenv("CI_REPOSITORY_URL",None))
82
83 # Verify user input. All data must be set.
84 if not all([x for x in [header["PRIVATE-TOKEN"], url, project]]):
85 # Issue a warning - but return gracefully
86 warnings.warn("Could not validate release CLI input parameters. One or all of <token>, <base_url> and <project_url> are missing.", RuntimeWarning) #@UndefinedVariable
87 return 0
88
89 # Determine server and namespace from the given URL
90 server = '{uri.scheme}://{uri.netloc}/'.format(uri=urlparse(project))
91 namespace = os.path.splitext(project.split(server)[-1])[0]
92 server = str(2*posixpath.sep).join([server.split(posixpath.sep*2)[0], server.split("@")[-1].split(posixpath.sep*2)[-1]])
93
94 # Release asset options
95 options = kwargs.get("assets",{"generic":{"tar.gz":"Source code (tar.gz)",
96 "zip":"Source code (zip)",
97 "whl":"Python installer (whl)",
98 "exe":"Standalone application (exe)"},
99 "pypi":{"whl":"PyPi package (whl)"}})
100
101 # Could not determine all required values. Skipping.
102 if not all([x for x in [url, server, namespace]]): pass
103
104 # This is the current repository
105 repository = namespace.split(posixpath.sep)[-1]
106
107 # This is the project group
108 group_url = posixpath.dirname(posixpath.join(server, "groups",namespace))
109
110 # Search for the given namespace in all groups
111 found = [group
112 for page in range(1,int(requests.head(posixpath.join(url,"groups"), headers=header).headers['x-total-pages']) + 1)
113 for group in requests.get(posixpath.join(url,"groups"), params={"page":str(page)}, headers=header).json()
114 if group_url in group["web_url"]]
115
116 # We found the group
117 if found: found = [group for group in found if len(group["web_url"]) == len(group_url)][0]
118 else: raise RuntimeError("Could not determine group of current project.")
119
120 ## Search for the given project within the group
121 # Use case insensitive search
122 found = [project
123 # Iterate through all pages
124 for page in range(1,int(requests.head(posixpath.join(url,"groups",str(found["id"]),"projects"), headers=header).headers['x-total-pages']) + 1)
125 # Iterate through all projects within the current group
126 for project in requests.get(posixpath.join(url,"groups",str(found["id"]),"projects"), params={"page":str(page)}, headers=header).json()
127 # Collect all entries where repository matches name or path
128 if any(repository.lower() in match.lower() for match in [project["name"],project["path"]] ) ][0]
129 # Search for the latest release
130 releases = [release
131 for page in range(1,int(requests.head(posixpath.join(url,"projects",str(found["id"]),"releases"), headers=header).headers.get('x-total-pages',0)) + 1)
132 for release in requests.get(posixpath.join(url,"projects",str(found["id"]),"releases"), params={"page":str(page)}, headers=header).json()]
133
134 # This is the latest release. Can be empty
135 versions = [d["tag_name"][1:] for d in releases]
136 # Allow modern naming convention like releases
137 try: versions.sort(key=StrictVersion)
138 except: versions.sort(key=LooseVersion)
139
140 # Check if an older version is found. However, versions can be empty. Acknowledge that
141 if not versions:
142 try: tag = kwargs.get("tag",os.getenv("TAG"))
143 except: raise RuntimeError("Could not determine project tag. Please create an environment variable explicitly named 'TAG'.")
144 else: tag = kwargs.get("tag",os.getenv("TAG","v%s" % versions[-1]))
145
146 # Search for all available packages. A package must have the same tag as the release
147 package = [package
148 for page in range(1,int(requests.head(posixpath.join(url,"projects",str(found["id"]),"packages"), headers=header).headers['x-total-pages']) + 1)
149 for package in requests.get(posixpath.join(url,"projects",str(found["id"]),"packages"), params={"page":str(page)}, headers=header).json()
150 if version.parse(package["version"]) == version.parse(tag.split("v")[-1])]
151
152 # There is no package data for the requested release
153 if not package: raise RuntimeError("Could not determine latest package associated with tag: %s" % str(tag))
154
155 # Collect all assets
156 sha = [x["pipeline"]["sha"] for x in package if set([x["pipeline"]["project_id"],x["name"]]) == set([found["id"],found["name"]])]
157 assets = {x["package_type"]:posixpath.join(server,x["_links"]["web_path"][1:]) for x in package}
158
159 # Fetch all uploaded files within assets
160 files = [(key, x) for key, asset in assets.items()
161 for x in requests.get(posixpath.join(url,"projects",str(found["id"]),"packages",asset.split(posixpath.sep)[-1],"package_files"), headers=header).json() ]
162
163 # Construct download links
164 for key in list(assets.keys()): assets.update({key:{"url":posixpath.join(server,namespace,"-","package_files")}})
165 for key in list(assets.keys()): assets[key].update({"files": [{ x[-1]["file_name"] : x[-1]["id"] } for x in files if key in x[0] ]})
166
167 # Create a collection of all assets
168 collector = [{"name":name, "url": posixpath.join(assets[key]["url"],str(ID),"download")}
169 for key in options
170 for extension, name in options[key].items()
171 for entry in assets.get(key,{}).get("files",[])
172 for file, ID in entry.items() if file.endswith(extension)]
173
174 # Validate that only singular and correct entries remain
175 collector = [entry for entry in collector if requests.get(entry["url"]).status_code == 200]
176 collector = [{"name":key, "url":value} for key, value in {entry["name"]:entry["url"] for entry in collector}.items()]
177
178 # Create the request body
179 release = {"name":"Release of %s" % tag, "description":"Created using CLI", "tag_name":str(tag),"assets": {"links":collector}}
180
181 # Only meaningful if tagged version does not yet exists.
182 if sha: release.update({"ref":sha[0]})
183 elif os.getenv("CI_COMMIT_SHA",""): release.update({"ref":os.getenv("CI_COMMIT_SHA")})
184
185 # Delete outdated release. Defaults to True
186 if kwargs.get("silent_update",True): requests.delete(posixpath.join(url,"projects",str(found["id"]),"releases",tag), headers=header)
187
188 # Execute the release command
189 r = requests.post(posixpath.join(url,"projects",str(found["id"]),"releases"), json=release, headers=header)
190 try: r.raise_for_status()
191 except:
192 warnings.warn("Creating or updating the release raised an unexpected return code. Please verify the result.", RuntimeWarning) #@UndefinedVariable
193 warnings.warn("%s" % r.text, RuntimeWarning) #@UndefinedVariable
194 warnings.warn("%s" % str(release), RuntimeWarning) #@UndefinedVariable
195 # Return the response
196 return r
197
198def main(**kwargs):
199 """
200 Main command line parser.
201 """
202 settings = {}
203 options = {"choices":kwargs.pop("choices",["release","housekeeping"])}
204 commands = globals(); commands.update(kwargs.get("register",{}))
205 options["choices"] += list(kwargs.pop("register",{}).keys())
206 if not kwargs.get("method",""):
207 parser = argparse.ArgumentParser(description='CLI wrapper options for GitLab CI.')
208 parser.add_argument('method', metavar='namespace', type=str, nargs=1, choices=options["choices"],
209 help='An option identifier. Unknown arguments are ignored. Allowed values are: '+', '.join(options["choices"]))
210 # The options are equally valid for all commands
211 parser.add_argument('-b', '--base_url', type=str, nargs=1, help="Base API v4 URL of a GitLab instance. Defaults to GitLab instance of DLR.")
212 parser.add_argument('-t', '--token', type=str, nargs=1, help="Token to be used for authentication.")
213 parser.add_argument('-i', '--identifier', type=str, nargs=1, help="A valid, unique project identifier.")
214 parser.add_argument('-p', '--package', type=str, nargs=1, help="A valid project name.")
215 parser.add_argument('-v', '--version', type=str, nargs=1, help="A valid version identifier for PyPI.")
216 parser.add_argument("-o", '--output', type=str, nargs=1, help="Absolute path to output directory. Defaults to current project folder.")
217 # Select method
218 args, _ = parser.parse_known_args()
219 method = str(args.method[0])
220 # Find all given command line parameters
221 given = [x for x in vars(args) if getattr(args,x)]
222 # Add them to settings
223 settings.update({x:next(iter(getattr(args,x))) for x in given})
224 else: method = kwargs.get("method")
225 settings.update(copy.deepcopy(kwargs))
226 # Execute wrapper within this script or parse the command to git
227 if not method in globals(): raise RuntimeError("Unknown GitLab CI namespace option.")
228 else: commands[method](**settings)
229 pass
230
231if __name__ == "__main__":
232 main(); sys.exit()
main()
Provide a custom error handler for the interruption event triggered by this wrapper to prevent multip...
Definition __init__.py:62