def invocations()

in src/vw-serving/src/vw_serving/serve.py [0:0]
67 lines of code
13 McCabe index (conditional complexity)

def invocations():
    content_type, content_parameters = ScoringService.parse_content_type(flask.request.content_type)
    response_content_type = flask.request.headers.get("Accept", "application/json")
    if content_type not in [CONTENT_TYPE_JSON, CONTENT_TYPE_JSONLINES, CONTENT_TYPE_CSV]:
        sdk_error = InferenceCustomerError("Content-type {} not supported".format(content_type))
        ScoringService._report_sdk_error(sdk_error)
        return flask.Response(
            response="content-type {} not supported".format(content_type), status=httplib.UNSUPPORTED_MEDIA_TYPE
        )

    payload = flask.request.data
    if len(payload) == 0:
        return flask.Response(response="", status=httplib.NO_CONTENT)

    try:
        model = ScoringService.get_model()
    except Exception as e:
        sdk_error = convert_to_algorithm_error(e)
        ScoringService._report_sdk_error(sdk_error)
        return flask.Response(response="unable to load model", status=httplib.INTERNAL_SERVER_ERROR,
                              mimetype="application/json", content_type="application/json")

    if content_type == CONTENT_TYPE_JSON:
        data = json.loads(payload.decode("utf-8"))

        request_type = data.get("request_type", "observation").lower()

        if request_type == "reward":
            event_id = data["event_id"]
            reward = data["reward"]
            blob_to_log = {"event_id": int(event_id), "reward": float(reward), "type": "rewards"}
            blob_to_log = json.dumps(blob_to_log)
            if ScoringService.LOG_INFERENCE_DATA:
                ScoringService._redis_client.publish(REDIS_PUBLISHER_CHANNEL, blob_to_log)
                status = "success"
            else:
                status = "failure"
            return flask.Response(response='{"status": "%s"}' % status, status=httplib.OK)

        elif request_type == "model_id":
            model_info_payload = json.dumps({"model_id": ScoringService._model_id,
                                             "soft_model_update_status": "TBD: To be used for indicating rollbacks"})
            return flask.Response(response=model_info_payload, status=httplib.OK, mimetype="application/json",
                                  content_type="application/json")

        else:
            observation = data["observation"]

            response_payload = _score_json(model, observation)
            return flask.Response(response=response_payload, status=httplib.OK, mimetype="application/json",
                                  content_type="application/json")
    elif content_type == CONTENT_TYPE_JSONLINES:
        #  Content type is application/jsonlines, which means this is Batch Inference mode
        data = payload.decode("utf-8")
        f = StringIO(data)
        response = [_score_json(model, json.loads(line), response_content_type=response_content_type) for line in f.readlines()]
        response_payload = "\n".join(response)
        return flask.Response(response=response_payload, status=httplib.OK, mimetype=response_content_type,
                              content_type=response_content_type)
    else:
        # content type is csv, batch inference
        data = payload.decode("utf-8")
        rows_ = np.genfromtxt(StringIO(data), delimiter=',').astype(float)
        rows = rows_.tolist()
        # if this was a single record, we will just have a single list
        # the loop below expects a list of lists, so pack it up
        if len(rows_.shape) == 1:
            rows = [rows]
        response = [
            _score_json(
                model,
                row,
                response_content_type=response_content_type
            )
            for row in rows
        ]
        response_payload = "\n".join(response)
        return flask.Response(response=response_payload, status=httplib.OK, mimetype=response_content_type,
                              content_type=response_content_type)