build.lib.utils: Add module

Add some utility functions, namely:

  run_cmd(cmd: list[str], wd=None, throw=True, verbose=False) -> str|None
  run_curl(args: list[str], parse_json: bool=True) -> dict|str
  get_username(args: Namespace|None=None, url: str|None=None) -> str
  get_password(args: Namespace|None=None, url: str|None=None, askpass_env: list[str]=[]) -> str

Signed-off-by: Jan Lindemann <jan@janware.com>

src/python/jw/build/lib/util.py

@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+
+import os, sys, subprocess, json, time
+from argparse import Namespace
+from urllib.parse import urlparse
+
+def run_cmd(cmd: list[str], wd=None, throw=True, verbose=False) -> str|None: # export
+
+    tokens = [cmd[0]]
+    for token in cmd[1:]:
+        if token.find(' ') != -1:
+            token = '"' + token + '"'
+        tokens.append(token)
+    subject = ' '.join(tokens)
+    if wd is not None:
+        subject += f' in {wd}'
+    if verbose:
+        delim_len = 120
+        delim = f'---- running {subject} -'
+        delim = delim + '-' * (delim_len - len(delim))
+        print(',' + delim + ' >')
+
+    cwd: str|None = None
+    if wd is not None:
+        cwd = os.getcwd()
+        os.chdir(wd)
+
+    ret = ''
+    try:
+        p = subprocess.Popen(cmd, shell=False, stdout=subprocess.PIPE, stderr=None, close_fds=True)
+        for line in iter(p.stdout.readline, b''):
+            line = line.decode(sys.stdout.encoding)
+            ret += line
+        p.wait()
+        if verbose:
+            print('`' + delim + ' <')
+        if p.returncode:
+            if verbose:
+                print(' '.join(cmd) + ' failed')
+            raise Exception(time.strftime('%Y-%m-%d %H:%M') + f': failed to run "{subject}"')
+    finally:
+        if cwd:
+            os.chdir(cwd)
+    return ret
+
+def run_curl(args: list[str], parse_json: bool=True, wd=None, throw=None, verbose=False) -> dict|str: # export
+    cmd = ['curl']
+    if not verbose:
+        cmd.append('-s')
+    cmd.extend(args)
+    ret = run_cmd(cmd, wd=wd, throw=throw, verbose=verbose)
+    if parse_json:
+        return json.loads(ret)
+    return ret
+
+def get_username(args: Namespace|None=None, url: str|None=None) -> str: # export
+    url_user = None if url is None else urlparse(url).username
+    if args is not None:
+        if args.username is not None:
+            if url_user is not None and url_user != args.username:
+                raise Exception(f'Username mismatch: called with --username="{args.username}", URL has user name "{url_user}"')
+            return args.username
+    if url is not None:
+        return url_user
+    raise Exception(f'Neither URL nor command-line arguments available, can\'t get user name')
+
+def get_password(args: Namespace|None=None, url: str|None=None, askpass_env: list[str]=[]) -> str: # export
+    if args is None and url is None and not askpass_env:
+        raise Exception(f'Neither URL nor command-line arguments nor askpass environment variable available, can\'t get password')
+    if args is not None and hasattr(args, 'password'): # use getattr(), because we don't necessarily want to have insecure --password among options
+        ret = getattr(args, 'password')
+        if ret is not None:
+            return ret
+    if url is not None:
+        parsed = urlparse(url)
+        if parsed.password is not None:
+            return parsed.password
+    exes = []
+    if args is not None:
+        exes.append(getattr(args, 'askpass'))
+    for var in askpass_env:
+        exes.append(os.getenv(var))
+    for exe in exes:
+        if exe is None:
+            continue
+        ret = run_cmd(exe, throw=False)
+        if ret is not None:
+            return ret
+    return None