def batch_job_wes_state()

in packages/cdk/lib/wes_adapter/amazon_genomics/wes/adapters/BatchAdapter.py [0:0]

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

    def batch_job_wes_state(job_status, job_status_reason) -> State:
        """
        Derive WES job state from AWS Batch job description (of workflow job)
        :param job_status: Job status as given by AWS Batch.
            See: https://docs.aws.amazon.com/batch/latest/userguide/job_states.html
        :param job_status_reason: Status reason used to distinguish CANCELED from FAILED run status for WES
            See: https://docs.aws.amazon.com/batch/latest/APIReference/API_JobSummary.html
        :return: WES Job status equivalent.
            See: https://ga4gh.github.io/workflow-execution-service-schemas/docs/#operation/GetRunStatus
        """
        #
        if job_status in ("SUBMITTED", "PENDING", "RUNNABLE"):
            return State.QUEUED
        elif job_status == "STARTING":
            return State.INITIALIZING
        elif job_status == "RUNNING":
            return State.RUNNING
        elif job_status == "SUCCEEDED":
            return State.COMPLETE
        elif job_status == "FAILED":
            if job_status_reason == USER_CANCELLATION_REASON:
                return State.CANCELED
            # TODO: detect SYSTEM_ERROR (probably, anything but "Essential container in task exited")
            return State.EXECUTOR_ERROR
        else:
            return State.UNKNOWN