def invocations()

in src/sagemaker_xgboost_container/algorithm_mode/serve.py [0:0]

• 35 lines of code
• 9 McCabe index (conditional complexity)

def invocations():
    payload = flask.request.data
    if len(payload) == 0:
        return flask.Response(response="", status=http.client.NO_CONTENT)

    try:
        dtest, content_type = serve_utils.parse_content_data(payload, flask.request.content_type)
    except Exception as e:
        logging.exception(e)
        return flask.Response(response=str(e), status=http.client.UNSUPPORTED_MEDIA_TYPE)

    try:
        format = load_model()
    except Exception as e:
        logging.exception(e)
        return flask.Response(response="Unable to load model: %s" % e, status=http.client.INTERNAL_SERVER_ERROR)

    try:
        preds = ScoringService.predict(data=dtest, content_type=content_type, model_format=format)
    except Exception as e:
        logging.exception(e)
        return flask.Response(response="Unable to evaluate payload provided: %s" % e, status=http.client.BAD_REQUEST)

    try:
        accept = _parse_accept(flask.request)
    except Exception as e:
        logging.exception(e)
        return flask.Response(response=str(e), status=http.client.NOT_ACCEPTABLE)

    if serve_utils.is_selectable_inference_output():
        return _handle_selectable_inference_response(preds, accept)

    preds_list = preds.tolist()
    if SAGEMAKER_BATCH:
        return_data = "\n".join(map(str, preds_list)) + "\n"
    else:
        if accept == _content_types.JSON:
            return_data = serve_utils.encode_predictions_as_json(preds_list)
        else:
            return_data = encoders.encode(preds_list, accept)

    return flask.Response(response=return_data, status=http.client.OK, mimetype=accept)