def from_describe_response()

in src/sagemaker/remote_function/client.py [0:0]

• 62 lines of code
• 15 McCabe index (conditional complexity)

    def from_describe_response(describe_training_job_response, sagemaker_session):
        """Construct a Future from a describe_training_job_response object."""
        future = Future()
        job_exception = None
        client_exception = None
        job_return = None
        job = _Job.from_describe_response(describe_training_job_response, sagemaker_session)
        if describe_training_job_response["TrainingJobStatus"] in ["Stopping", "Stopped"]:
            state = _CANCELLED
        elif describe_training_job_response["TrainingJobStatus"] == "Completed":
            state = _FINISHED
            try:
                job_return = serialization.deserialize_obj_from_s3(
                    sagemaker_session=sagemaker_session,
                    s3_uri=s3_path_join(job.s3_uri, RESULTS_FOLDER),
                    hmac_key=job.hmac_key,
                )
            except DeserializationError as e:
                client_exception = e
            except ServiceError as e:
                client_exception = e
        elif describe_training_job_response["TrainingJobStatus"] == "Failed":
            state = _FINISHED
            try:
                job_exception = serialization.deserialize_exception_from_s3(
                    sagemaker_session=sagemaker_session,
                    s3_uri=s3_path_join(job.s3_uri, EXCEPTION_FOLDER),
                    hmac_key=job.hmac_key,
                )
            except ServiceError as serr:
                chained_e = serr.__cause__
                if (
                    isinstance(chained_e, ClientError)
                    and chained_e.response["Error"]["Code"] == "404"  # pylint: disable=no-member
                    and chained_e.response["Error"]["Message"]  # pylint: disable=no-member
                    == "Not Found"
                ):
                    if (
                        "FailureReason" in describe_training_job_response
                        and describe_training_job_response["FailureReason"]
                        and "RuntimeEnvironmentError: "
                        in describe_training_job_response["FailureReason"]
                    ):
                        failure_msg = describe_training_job_response["FailureReason"].replace(
                            "RuntimeEnvironmentError: ", ""
                        )
                        job_exception = RuntimeEnvironmentError(failure_msg)
                    else:
                        job_exception = RemoteFunctionError(
                            "Failed to execute remote function. "
                            + "Check corresponding job for details."
                        )
                else:
                    job_exception = serr
            except DeserializationError as e:
                client_exception = e
        else:
            state = _RUNNING

        future._job = job
        future._state = state
        future._exception = job_exception or client_exception
        future._return = job_return
        return future