def update_target()

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

• 57 lines of code
• 13 McCabe index (conditional complexity)

def update_target():  # pylint: disable=too-many-return-statements,too-many-branches
    """Use terraform to update a target."""
    token_dict = get_token.get_token(flask.request, token_type="access_token")
    if "response" in token_dict:
        return token_dict["response"]
    if not flask.request.args.get("project_id"):
        return flask.Response(
            status=200,
            response=json.dumps({"status": "BLOCKED", "reason": "NO_PROJECT_ID"}),
        )
    response = asu.validate_project_id(
        flask.request.args["project_id"], token_dict["access_token"]
    )
    if response:
        return response
    content = flask.request.get_json(silent=True)
    access_token = token_dict["access_token"]

    env = asu.get_terraform_env(
        access_token,
        flask.request.args,
        debug=asu.get_debug(flask.request),
    )
    if "response" in env:
        return env["response"]
    targets = content.get("targets")
    destroy = content["destroy"]

    if targets == ["all"]:
        update_targets = None
    else:
        update_targets = targets

    ctx = context.Context()
    module = "/deploy/terraform/main.tf"
    prefix = f'terraform/{flask.request.args["project_id"]}'
    with tempfile.TemporaryDirectory() as workdir:
        result = asu.tf_init(ctx, module, workdir, env, prefix)
        if result:
            return result

        if update_targets:
            for target in update_targets:
                result = asu.tf_plan(ctx, module, workdir, env, target=target)
                if result and "response" in result:
                    return result["response"]
                result = asu.tf_apply(ctx, module, workdir, env, destroy, target=target)
        else:
            result = asu.tf_plan(ctx, module, workdir, env)
            result = asu.tf_apply(ctx, module, workdir, env, destroy)
        if result:
            return result

        result = asu.tf_state_list(ctx, module, workdir, env)
        if "response" in result:
            return result["response"]

        response = flask.Response(
            status=200,
            response=json.dumps({"status": "OK", "resources": result["resources"]}),
        )
        return au.register_action(
            flask.request, response, au.ACTIONS.UPDATE_STATUS, {"service": "ingress"}
        )