def _validate_inputs()

in src/sagemaker/workflow/notebook_job_step.py [0:0]

• 40 lines of code
• 18 McCabe index (conditional complexity)

    def _validate_inputs(self):
        """Validation function for the notebook job step inputs."""

        errors = []
        # notebook job name should start with letters and contain only letters, numbers, hyphens,
        # and underscores
        if not self.notebook_job_name or not bool(
            re.match(r"^[a-zA-Z][a-zA-Z0-9_-]*$", self.notebook_job_name)
        ):
            errors.append(
                f"Notebook Job Name({self.notebook_job_name}) is not valid. Valid name "
                f"should start with letters and contain only letters, numbers, hyphens, "
                f"and underscores."
            )

        # input notebook is required
        if not self.input_notebook or not os.path.isfile(self.input_notebook):
            errors.append(
                f"The required input notebook({self.input_notebook}) is not a valid " f"file."
            )

        # init script is optional
        if self.initialization_script and not os.path.isfile(self.initialization_script):
            errors.append(f"The initialization script({self.input_notebook}) is not a valid file.")

        if self.additional_dependencies:
            for path in self.additional_dependencies:
                if not os.path.exists(path):
                    errors.append(
                        f"The path({path}) specified in additional dependencies does not exist."
                    )
        # image uri is required
        if not self.image_uri or self._region_from_session not in self.image_uri:
            errors.append(
                f"The image uri(specified as {self.image_uri}) is required and "
                f"should be hosted in same region of the session"
                f"({self._region_from_session})."
            )

        if not self.kernel_name:
            errors.append("The kernel name is required.")

        # validate the role after resolving.
        if not self.role:
            errors.append(f"The IAM role is '{self.role}' and no default role can be found.")

        # validate the s3 root uri after resolving
        if not self.s3_root_uri:
            errors.append(
                f"The s3_root_uri is '{self.s3_root_uri}' and no default s3_root_uri can"
                f" be found."
            )

        if errors:
            errors_message = "\n  - ".join(errors)
            raise ValueError(f"Validation Errors: \n  - {errors_message}")