fix: handle empty inputs/outputs in ProcessingJob.from_processing_name() (#1348)

Author: knakad

src/sagemaker/processing.py

@@ -648,10 +648,9 @@ def from_processing_name(cls, sagemaker_session, processing_job_name):
         """
         job_desc = sagemaker_session.describe_processing_job(job_name=processing_job_name)
 
-        return cls(
-            sagemaker_session=sagemaker_session,
-            job_name=processing_job_name,
-            inputs=[
+        inputs = None
+        if job_desc.get("ProcessingInputs"):
+            inputs = [
                 ProcessingInput(
                     source=processing_input["S3Input"]["S3Uri"],
                     destination=processing_input["S3Input"]["LocalPath"],
@@ -664,19 +663,31 @@ def from_processing_name(cls, sagemaker_session, processing_job_name):
                     s3_compression_type=processing_input["S3Input"].get("S3CompressionType"),
                 )
                 for processing_input in job_desc["ProcessingInputs"]
-            ],
-            outputs=[
+            ]
+
+        outputs = None
+        if job_desc.get("ProcessingOutputConfig") and job_desc["ProcessingOutputConfig"].get(
+            "Outputs"
+        ):
+            outputs = [
                 ProcessingOutput(
-                    source=job_desc["ProcessingOutputConfig"]["Outputs"][0]["S3Output"][
-                        "LocalPath"
-                    ],
-                    destination=job_desc["ProcessingOutputConfig"]["Outputs"][0]["S3Output"][
-                        "S3Uri"
-                    ],
-                    output_name=job_desc["ProcessingOutputConfig"]["Outputs"][0]["OutputName"],
+                    source=processing_output["S3Output"]["LocalPath"],
+                    destination=processing_output["S3Output"]["S3Uri"],
+                    output_name=processing_output["OutputName"],
                 )
-            ],
-            output_kms_key=job_desc["ProcessingOutputConfig"].get("KmsKeyId"),
+                for processing_output in job_desc["ProcessingOutputConfig"]["Outputs"]
+            ]
+
+        output_kms_key = None
+        if job_desc.get("ProcessingOutputConfig"):
+            output_kms_key = job_desc["ProcessingOutputConfig"].get("KmsKeyId")
+
+        return cls(
+            sagemaker_session=sagemaker_session,
+            job_name=processing_job_name,
+            inputs=inputs,
+            outputs=outputs,
+            output_kms_key=output_kms_key,
         )
 
     @classmethod

tests/integ/test_processing.py

@@ -20,7 +20,13 @@
 from sagemaker import Session
 from sagemaker.fw_registry import default_framework_uri
 
-from sagemaker.processing import ProcessingInput, ProcessingOutput, ScriptProcessor, Processor
+from sagemaker.processing import (
+    ProcessingInput,
+    ProcessingOutput,
+    ScriptProcessor,
+    Processor,
+    ProcessingJob,
+)
 from sagemaker.sklearn.processing import SKLearnProcessor
 from sagemaker.utils import sts_regional_endpoint
 from tests.integ import DATA_DIR
@@ -475,6 +481,37 @@ def test_script_processor_with_no_inputs_or_outputs(
 
     assert job_description["StoppingCondition"] == {"MaxRuntimeInSeconds": 3600}
 
+    job_from_name = ProcessingJob.from_processing_name(
+        sagemaker_session=sagemaker_session,
+        processing_job_name=job_description["ProcessingJobName"],
+    )
+    job_description = job_from_name.describe()
+
+    assert job_description["ProcessingInputs"][0]["InputName"] == "code"
+
+    assert job_description["ProcessingJobName"].startswith("test-script-processor-with-no-inputs")
+
+    assert job_description["ProcessingJobStatus"] == "Completed"
+
+    assert job_description["ProcessingResources"]["ClusterConfig"]["InstanceCount"] == 1
+    assert (
+        job_description["ProcessingResources"]["ClusterConfig"]["InstanceType"] == cpu_instance_type
+    )
+    assert job_description["ProcessingResources"]["ClusterConfig"]["VolumeSizeInGB"] == 100
+
+    assert job_description["AppSpecification"]["ContainerArguments"] == ["-v"]
+    assert job_description["AppSpecification"]["ContainerEntrypoint"] == [
+        "python3",
+        "/opt/ml/processing/input/code/dummy_script.py",
+    ]
+    assert job_description["AppSpecification"]["ImageUri"] == image_uri
+
+    assert job_description["Environment"] == {"DUMMY_ENVIRONMENT_VARIABLE": "dummy-value"}
+
+    assert ROLE in job_description["RoleArn"]
+
+    assert job_description["StoppingCondition"] == {"MaxRuntimeInSeconds": 3600}
+
 
 @pytest.mark.canary_quick
 def test_processor(sagemaker_session, image_uri, cpu_instance_type, output_kms_key):