python popen example

example.py

stdout = lambda x: sys.stdout.write(x + '\n')
stderr = lambda x: sys.stderr.write(x + '\n')


def sanitize_output(output):
    output = re.sub(
        r"[\"']?password[\"']?: ['\"a-z0-9A-Z]+", "'password': '###'", output
    )
    output = re.sub(
        r"postgres:\/\/avnadmin:[a-z0-9A-Z]+@", "postgres://admin:###@", output
    )
    tok = os.environ.get("AIVEN_AUTH_TOKEN")
    if tok:
        output = output.replace(tok, "###")
    return output


def do_popen(
    cmd: str,
    err_msg: AnyStr = "",
    _json: bool = None,
    exc=Exception,
    no_buffer=False,
    quiet: bool = False,
) -> Union[str, dict]:
    cmd = cmd.split()
    if no_buffer:
        process = Popen(
            cmd,
            stderr=PIPE,
            stdout=PIPE,
            env=os.environ,
            bufsize=1,
            universal_newlines=True,
        )
        lines = []
        with process.stderr:
            for line in process.stderr:
                stdout(line)
                lines.append(line)
        process.wait()
        return ""
    else:
        process = Popen(cmd, stderr=PIPE, stdout=PIPE, env=os.environ)
        _stdout, _stderr = process.communicate()
    errcode = process.returncode
    if quiet:
        return errcode
    if errcode:
        stderr(sanitize_output(" ".join(cmd + [_stderr.decode("utf8")])))
        raise exc(err_msg)
    stdout(sanitize_output(_stdout.decode("utf8")))
    if _json:
        return json.loads(_stdout.decode("utf8"))
    return sanitize_output(_stdout.decode("utf8"))

list_cmd = """avn --auth-token {auth_token} service list --project {project} {app_name} --json""".format

config = {
    "auth_token": f'"{os.environ.get("AIVEN_AUTH_TOKEN")}"',  # set in heroku staging env vars in dashboard "reveal config vars", and aiven console
    "app_name": f"{os.environ.get('HEROKU_APP_NAME', ''.join(random.choices(string.ascii_lowercase, k=14)))}",
    "project": os.environ.get("AIVEN_PROJECT_NAME", "propertymeld-f3df"),
}

service_config = {
    "cloud": "do-nyc",
    "service_type": "pg",
    "plan": "startup-4",  # hobbyist does not support pooling
    "pg_version": "pg_version=11",
}


do_popen(
    list_cmd(**config),
    err_msg=f"Had difficulties waiting: {config.get('app_name')}",
    no_buffer=True,
)