def asset_status()

in dialogflow-cx/vpc-sc-demo/backend/asset_blueprint.py [0:0]
67 lines of code
14 McCabe index (conditional complexity)

def asset_status():  # pylint: disable=too-many-locals,too-many-return-statements,too-many-branches
    """Get status of terraform-tracked assets."""
    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

    target = flask.request.args.get("target", None)
    update = flask.request.args.get("update", True)
    env = asu.get_terraform_env(
        token_dict["access_token"],
        flask.request.args,
        debug=asu.get_debug(flask.request),
    )
    if "response" in env:
        return env["response"]
    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

        resource_id_dict = {}
        if update:
            result = asu.tf_plan(ctx, module, workdir, env, target=target)
            if result is not None:
                if "response" in result:
                    return result["response"]
                for hook in result["hooks"]["refresh_complete"]:
                    if "id_value" in hook:
                        resource_id_dict[hook["resource"]["addr"]] = hook["id_value"]

        if ACCESS_POLICY_RESOURCE in resource_id_dict:
            access_policy_id = resource_id_dict[ACCESS_POLICY_RESOURCE]
            response = asu.get_access_policy_title(
                token_dict["access_token"], access_policy_id
            )
            if "response" in response:
                return response["response"]
            access_policy_title = response["access_policy_title"]
        else:
            access_policy_title = None

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

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