def tf_plan()

in dialogflow-cx/vpc-sc-demo/backend/asset_utilities.py [0:0]

• 53 lines of code
• 10 McCabe index (conditional complexity)

def tf_plan(context, module, workdir, env, target=None):
    """Create terraform plan."""
    debug = "TF_LOG" in env
    target_option = f"-target={target}" if target else ""
    json_option = "-json" if not debug else ""
    promise = context.run(
        (
            f"cp {module} {workdir} && "
            f'terraform -chdir="{workdir}" plan {json_option} '
            "-refresh-only -var "
            f'access_token=\'{env["GOOGLE_OAUTH_ACCESS_TOKEN"]}\' {target_option}'
        ),
        warn=True,
        hide=True,
        asynchronous=True,
        env=env,
    )
    result = promise.join()

    if debug:
        logging.debug(result.exited)
        logging.debug(result.stdout)
        logging.debug(result.stderr)
    else:
        errors = []
        hooks = {
            "refresh_start": [],
            "refresh_complete": [],
            "apply_complete": [],
            "apply_start": [],
        }
        lines = result.stdout.split("\n")
        for line in lines:
            if line.strip():
                try:
                    message = json.loads(line.strip())
                    if message["@level"] == "error":
                        errors.append(message)
                    if "hook" in message:
                        hooks[message["type"]].append(message["hook"])
                except KeyError:
                    logging.debug("COULD NOT LOAD: %s", line)
        if errors:
            return {
                "response": Response(
                    status=500,
                    response=json.dumps(
                        {
                            "status": "ERROR",
                            "errors": errors,
                        }
                    ),
                )
            }
        return {"hooks": hooks}