feat: Support local mode for Amazon SageMaker Processing jobs

src/sagemaker/local/entities.py

@@ -38,6 +38,135 @@
 HEALTH_CHECK_TIMEOUT_LIMIT = 120
 
 
+class _LocalProcessingJob(object):
+    """Placeholder docstring"""
+
+    _STARTING = "Starting"
+    _PROCESSING = "Processing"
+    _COMPLETED = "Completed"
+    _states = ["Starting", "Processing", "Completed"]
+
+    def __init__(self, container):
+        """
+        Args:
+            container:
+        """
+        self.container = container
+        self.state = "Created"
+        self.start_time = None
+        self.end_time = None
+        self.processing_job_name = ""
+        self.processing_inputs = None
+        self.processing_output_config = None
+        self.environment = None
+
+    def start(self, processing_inputs, processing_output_config, environment, processing_job_name):
+        """
+        Args:
+            processing_inputs:
+            processing_output_config:
+            environment:
+            processing_job_name:
+        """
+
+        for item in processing_inputs:
+            if item["S3Input"]:
+                data_uri = item["S3Input"]["S3Uri"]
+            else:
+                raise ValueError("Processing input must have a valid ['S3Input']")
+
+            if item["S3Input"]["S3InputMode"]:
+                input_mode = item["S3Input"]["S3InputMode"]
+            else:
+                raise ValueError("Processing input must have a valid ['S3InputMode']")
+
+            item["DataUri"] = data_uri
+
+            if (
+                "S3DataDistributionType" in item["S3Input"]
+                and item["S3Input"]["S3DataDistributionType"] != "FullyReplicated"
+            ):
+
+                raise RuntimeError(
+                    "DataDistribution: %s is not currently supported in Local Mode"
+                    % item["S3Input"]["S3DataDistributionType"]
+                )
+
+            if input_mode != "File":
+                raise RuntimeError(
+                    "S3InputMode: %s is not currently supported in Local Mode" % input_mode
+                )
+
+            if (
+                "S3CompressionType" in item["S3Input"]
+                and item["S3Input"]["S3CompressionType"] != "None"
+            ):
+
+                raise RuntimeError(
+                    "CompressionType: %s is not currently supported in Local Mode"
+                    % item["S3Input"]["S3CompressionType"]
+                )
+
+        if processing_output_config and "Outputs" in processing_output_config:
+            processing_outputs = processing_output_config["Outputs"]
+
+            for item in processing_outputs:
+                if item["S3Output"]:
+                    upload_mode = item["S3Output"]["S3UploadMode"]
+                else:
+                    raise ValueError("Processing output must have a valid ['S3Output']")
+
+                if upload_mode != "EndOfJob":
+                    raise RuntimeError(
+                        "UploadMode: %s is not currently supported in Local Mode." % upload_mode
+                    )
+
+        self.start_time = datetime.datetime.now()
+        self.state = self._PROCESSING
+
+        self.processing_job_name = processing_job_name
+        self.processing_inputs = processing_inputs
+        self.processing_output_config = processing_output_config
+        self.environment = environment
+
+        self.container.process(
+            processing_inputs, processing_output_config, environment, processing_job_name
+        )
+        self.end_time = datetime.datetime.now()
+        self.state = self._COMPLETED
+
+    def describe(self):
+        """Placeholder docstring"""
+
+        response = {
+            "ProcessingJobArn": self.processing_job_name,
+            "ProcessingJobName": self.processing_job_name,
+            "AppSpecification": {
+                "ImageUri": self.container.image,
+                "ContainerEntrypoint": self.container.container_entrypoint,
+                "ContainerArguments": self.container.container_arguments,
+            },
+            "Environment": self.environment,
+            "ProcessingInputs": self.processing_inputs,
+            "ProcessingOutputConfig": self.processing_output_config,
+            "ProcessingResources": {
+                "ClusterConfig": {
+                    "InstanceCount": self.container.instance_count,
+                    "InstanceType": self.container.instance_type,
+                    "VolumeSizeInGB": 30,
+                    "VolumeKmsKeyId": None,
+                }
+            },
+            "RoleArn": "<no_role>",
+            "StoppingCondition": {"MaxRuntimeInSeconds": 86400},
+            "ProcessingJobStatus": self.state,
+            "ProcessingStartTime": self.start_time,
+            "ProcessingEndTime": self.end_time,
+        }
+
+        return response
+
+
 class _LocalTrainingJob(object):
     """Placeholder docstring"""