utils.common

  1#!/usr/bin/env python3
  2"""
  3common
  4
  5Common code useful for any pipeline
  6
  7Example Usage:
  8    from utils import common
  9    from utils.common import subprocess_long
 10"""
 11
 12from subprocess_tee import run
 13import os
 14import re
 15import subprocess
 16from typing import TYPE_CHECKING, Any, List, Union
 17try:
 18    import loggy
 19except ImportError:
 20    from utils import loggy
 21
 22if TYPE_CHECKING:
 23    CompletedProcess = subprocess.CompletedProcess[Any]  # pylint: disable=E1136
 24else:
 25    CompletedProcess = subprocess.CompletedProcess
 26
 27try:
 28    from shlex import join  # type: ignore
 29except ImportError:
 30    from subprocess import list2cmdline as join  # pylint: disable=ungrouped-imports
 31
 32
 33class ChDir(object):
 34    """
 35    Step into a directory Temporarily
 36
 37    Usage Example:
 38        with ChDir("cdk"):
 39            loggy.info("I'm in the cdk folder")
 40    """
 41
 42    def __init__(self, path):
 43        self.old_dir = os.getcwd()
 44        self.new_dir = path if len(path) > 0 else os.getcwd()
 45
 46    def __enter__(self):
 47        os.chdir(self.new_dir)
 48
 49    def __exit__(self, *args):
 50        os.chdir(self.old_dir)
 51
 52
 53def removeprefix(s: str, prefix: str, /) -> str:
 54    """
 55    removeprefix()
 56
 57    This is built-in to string functions python3.9 but we use 3.8 for now.
 58
 59    s: String to check
 60    prefix: String prefix to look for
 61
 62    Returns: String modified with prefix removed
 63    """
 64    if s.startswith(prefix):
 65        return s[len(prefix):]
 66    else:
 67        return s[:]
 68
 69
 70def removesuffix(s: str, suffix: str, /) -> str:
 71    """
 72    removesuffix()
 73
 74    This is built-in to string functions python3.9 but we use 3.8 for now.
 75
 76    s: String to check
 77    suffix: String suffix to look for
 78
 79    Returns: String modified with suffix removed
 80    """
 81    if suffix and s.endswith(suffix):
 82        return s[:-len(suffix)]
 83    else:
 84        return s[:]
 85
 86
 87def resolve_gocd_pipeline_variable(param):
 88    """
 89    resolve_gocd_pipeline_variable()
 90
 91    GoCD pipeline variables get passed in as a literal string.
 92    Quickly check, resolve and pass back the true environment variable content.
 93    The `literal string` could be a variable surrounded by characters.
 94
 95    Examples:
 96        * $ENV_NAME
 97        * ${ENV_NAME}
 98        * my_${ENV_NAME}
 99        * ${ENV_NAME}_rc
100        * hello${ENV_NAME}goodbye
101
102    param: String containing potential env variable
103
104    Returns: String containing resolved env variable or original param if it can't resolve
105    """
106    _param = None
107
108    if "${" in param and "}" in param:
109        if param.startswith("${") and param.endswith("}"):
110            _param = removeprefix(param, "${")
111            _param = removesuffix(_param, "}")
112            _param = os.environ.get(_param, None)
113        else:
114            m = re.search('\\$\\{(.+?)\\}', param)
115            if m:
116                found = m.group(1)
117                _param = os.environ.get(found, None)
118                _param = _param if _param is None else param.replace("${" + found + "}", _param)
119    elif param.startswith("$"):
120        _param = removeprefix(param, "$")
121        _param = os.environ.get(_param, None)
122
123    return _param if _param is not None else param
124
125
126def subprocess_long(args: Union[str, List[str]], timeout=None, delay=None, check=None, shell=None, env=None):
127    """
128    subprocess_long():
129
130    imports subprocess_tee
131
132    Run long-running commands that are wrapped into a continuous loggy.info statement to keep
133    pipelines from exiting on zero output back to the GoCD server.
134
135    * Usage: common.subprocess_long(cmd="make install", timeout=15, delay=5, check=True)
136
137    timeout defaults to 30 minutes
138    delay defaults to 10 seconds
139    check defaults to False, use returned output.returncode to check for failure.
140        (Set to True if you want this function to error on command failure instead)
141    shell defaults to False
142    env defaults to None
143    """
144    timeout = 30 if timeout is None else timeout
145    delay = 10 if delay is None else delay
146    check = False if check is None else check
147    shell = False if shell is None else shell
148
149    if isinstance(args, str):
150        cmd = args
151    else:
152        # run was called with a list instead of a single item but asyncio
153        # create_subprocess_shell requires command as a single string, so
154        # we need to convert it to string
155        cmd = join(args)
156
157    longRunningProcessStart = 'counter=0; while [ $counter -lt ' + \
158        str(timeout*60) + \
159        ' ]; do counter=$((counter+' + \
160        str(delay) + \
161        ')); sleep ' + \
162        str(delay) + \
163        '; echo "common.subprocess_long(): LONG RUNNING PROCESS ENABLED - Running for $counter of ' + \
164        str(timeout*60) + \
165        ' seconds."; done & export _LONG_RUNNING_PROCESS=$!; '
166
167    longRunningProcessStop = '; retcode=$?; kill $_LONG_RUNNING_PROCESS; exit $retcode'
168
169    _process_output = run(longRunningProcessStart + cmd + longRunningProcessStop, env=env, shell=shell, check=check)
170
171    # loggy.info(_process_output.stdout)
172    # loggy.info(_process_output.stderr)
173    # loggy.info(_process_output.returncode)
174    return _process_output
175
176
177def subprocess_run(args: Union[str, List[str]], **kwargs: Any):
178    """
179    subprocess_run():
180
181    Replace the default subprocess_tee.run with check=True as default
182
183    * Usage: common.subprocess_run("make install")
184    """
185
186    if isinstance(args, str):
187        cmd = args
188    else:
189        # run was called with a list instead of a single item but asyncio
190        # create_subprocess_shell requires command as a single string, so
191        # we need to convert it to string
192        cmd = join(args)
193
194    try:
195        my_kwargs = kwargs.copy()
196        my_kwargs['check'] = kwargs.get("check", True)
197        my_kwargs['shell'] = kwargs.get("shell", True)
198        _process_output = run(cmd, **my_kwargs)
199    except subprocess.CalledProcessError as e:
200        loggy.error(f"common.subprocess_run(): Error: {str(e)}")
201        if _process_output and _process_output.stderr:
202            loggy.error(f"common.subprocess_run(): Process STDERR: {_process_output.stderr}")
203
204        raise
205
206    return _process_output
207
208
209def remove_empty_from_dict(d):
210    """
211    remove_empty_from_dict()
212
213    Removes empty values and empty lists from a dictionary
214
215    d: dict to modify
216
217    Returns: dict modified
218    """
219
220    if type(d) is dict:
221        return dict((k, remove_empty_from_dict(v)) for k, v in d.items() if v and remove_empty_from_dict(v))
222    elif type(d) is list:
223        return [remove_empty_from_dict(v) for v in d if v and remove_empty_from_dict(v)]
224    else:
225        return d