def result()

in src/sagemaker/remote_function/client.py [0:0]
63 lines of code
16 McCabe index (conditional complexity)

    def result(self, timeout: float = None) -> Any:
        """Returns the SageMaker job result.

        This method waits for the SageMaker job created from the remote function execution to
        complete for up to the timeout value (if specified). If timeout is ``None``,
        this method will wait until the SageMaker job completes.

        Args:
            timeout (float): Timeout in seconds to wait until the job is completed. ``None`` by
              default.

        Returns:
            The Python object returned by the remote function.
        """
        try:
            self.wait(timeout)
        except UnexpectedStatusException:
            pass

        with self._condition:
            if self._state == _PENDING:
                raise RuntimeError()

            if self._state == _RUNNING:
                if self._job.describe()["TrainingJobStatus"] == "Completed":
                    self._return = serialization.deserialize_obj_from_s3(
                        sagemaker_session=self._job.sagemaker_session,
                        s3_uri=s3_path_join(self._job.s3_uri, RESULTS_FOLDER),
                        hmac_key=self._job.hmac_key,
                    )
                    self._state = _FINISHED
                    return self._return
                if self._job.describe()["TrainingJobStatus"] == "Failed":
                    try:
                        self._exception = serialization.deserialize_exception_from_s3(
                            sagemaker_session=self._job.sagemaker_session,
                            s3_uri=s3_path_join(self._job.s3_uri, EXCEPTION_FOLDER),
                            hmac_key=self._job.hmac_key,
                        )
                    except ServiceError as serr:
                        chained_e = serr.__cause__
                        if (
                            isinstance(chained_e, ClientError)
                            and chained_e.response["Error"]["Code"]  # pylint: disable=no-member
                            == "404"
                            and chained_e.response["Error"]["Message"]  # pylint: disable=no-member
                            == "Not Found"
                        ):
                            if (
                                "FailureReason" in self._job.describe()
                                and self._job.describe()["FailureReason"]
                                and "RuntimeEnvironmentError: "
                                in self._job.describe()["FailureReason"]
                            ):
                                failure_msg = self._job.describe()["FailureReason"].replace(
                                    "RuntimeEnvironmentError: ", ""
                                )
                                self._exception = RuntimeEnvironmentError(failure_msg)
                            else:
                                self._exception = RemoteFunctionError(
                                    "Failed to execute remote function. "
                                    + "Check corresponding job for details."
                                )
                        else:
                            self._exception = serr
                    self._state = _FINISHED
                elif self._job.describe()["TrainingJobStatus"] == "Stopped":
                    self._state = _CANCELLED
                    raise RemoteFunctionError("Job for remote function has been aborted.")
                else:
                    raise TimeoutError(
                        "Job for remote function timed out before reaching a termination status."
                    )

            if self._state == _FINISHED:
                if self._exception:
                    raise self._exception
                return self._return

            return None