def _handle_invocation_post()

in tensorflow/inference/docker/build_artifacts/sagemaker/python_service.py [0:0]

• 74 lines of code
• 14 McCabe index (conditional complexity)

    def _handle_invocation_post(self, req, res, model_name=None):
        if SAGEMAKER_MULTI_MODEL_ENABLED:
            if model_name:
                if self._gunicorn_workers > 1:
                    if model_name not in self._mme_tfs_instances_status or not self._check_pid(
                        self._mme_tfs_instances_status[model_name][0].pid
                    ):
                        with lock():
                            self._sync_local_mme_instance_status()
                        self._sync_model_handlers()

                if model_name not in self._mme_tfs_instances_status:
                    res.status = falcon.HTTP_404
                    res.body = json.dumps(
                        {"error": "Model {} is not loaded yet.".format(model_name)}
                    )
                    return
                else:
                    log.info("model name: {}".format(model_name))
                    rest_ports = [
                        status.rest_port for status in self._mme_tfs_instances_status[model_name]
                    ]
                    rest_port = self._pick_port(rest_ports)
                    log.info("rest port: {}".format(str(rest_port)))
                    grpc_ports = [
                        status.grpc_port for status in self._mme_tfs_instances_status[model_name]
                    ]
                    grpc_port = grpc_ports[rest_ports.index(rest_port)]
                    log.info("grpc port: {}".format(str(grpc_port)))
                    data, context = tfs_utils.parse_request(
                        req,
                        rest_port,
                        grpc_port,
                        self._tfs_default_model_name,
                        model_name=model_name,
                    )
            else:
                res.status = falcon.HTTP_400
                res.body = json.dumps({"error": "Invocation request does not contain model name."})
                return
        else:
            # Randomly pick port used for routing incoming request.
            grpc_port = self._pick_port(self._tfs_grpc_ports)
            rest_port = self._pick_port(self._tfs_rest_ports)
            data, context = tfs_utils.parse_request(
                req,
                rest_port,
                grpc_port,
                self._tfs_default_model_name,
                channel=self._channels[grpc_port],
            )

        try:
            res.status = falcon.HTTP_200
            handlers = self._handlers
            if SAGEMAKER_MULTI_MODEL_ENABLED and model_name in self.model_handlers:
                log.info(
                    "Model-specific inference script for the model {} exists, importing handlers.".format(
                        model_name
                    )
                )
                handlers = self.model_handlers[model_name]
            elif not self._default_handlers_enabled:
                log.info(
                    "Universal inference script exists at path {}, importing handlers.".format(
                        INFERENCE_SCRIPT_PATH
                    )
                )
            else:
                log.info(
                    "Model-specific inference script and universal inference script both do not exist, using default handlers."
                )
            res.body, res.content_type = handlers(data, context)
        except Exception as e:  # pylint: disable=broad-except
            log.exception("exception handling request: {}".format(e))
            res.status = falcon.HTTP_500
            res.body = json.dumps({"error": str(e)}).encode("utf-8")  # pylint: disable=E1101