fix: address review comments (iteration #1)

Author: aviruthen

sagemaker-core/src/sagemaker/core/processing.py

@@ -16,14 +16,14 @@
 data pre-processing, post-processing, feature engineering, data validation, and model evaluation,
 and interpretation on Amazon SageMaker.
 """
-from __future__ import absolute_import
+from __future__ import annotations
 
 import json
 import logging
 import os
 import pathlib
 import re
-from typing import Dict, List, Optional, Union
+from typing import Dict, List, Literal, Optional, Union
 import time
 from copy import copy
 from textwrap import dedent
@@ -84,42 +84,37 @@
 
 logger = logging.getLogger(__name__)
 
+DEFAULT_PROCESSING_INPUT_PATH = "/opt/ml/processing/input"
+
 
 def processing_input_from_local(
     input_name: str,
     local_path: str,
     destination: str,
-    s3_data_type: str = "S3Prefix",
-    s3_input_mode: str = "File",
+    s3_data_type: Literal["S3Prefix", "ManifestFile"] = "S3Prefix",
+    s3_input_mode: Literal["File", "Pipe"] = "File",
 ) -> ProcessingInput:
-    """Creates a ProcessingInput from a local file or directory path.
+    """Create a ProcessingInput from a local file or directory path.
 
     This is a convenience factory that makes it clear users can pass local
     paths as processing job inputs. The local path is stored in
     ``ProcessingS3Input.s3_uri`` and will be automatically uploaded to S3
     when the processor's ``run()`` method is called.
 
-    Args:
-        input_name: The name for this processing input.
-        local_path: A local file or directory path to use as input.
-            This will be uploaded to S3 automatically before the
-            processing job starts.
-        destination: The container path where the input data will be
-            made available (e.g. ``/opt/ml/processing/input/data``).
-        s3_data_type: The S3 data type. Valid values: ``'S3Prefix'``,
-            ``'ManifestFile'`` (default: ``'S3Prefix'``).
-        s3_input_mode: The input mode. Valid values: ``'File'``,
-            ``'Pipe'`` (default: ``'File'``).
-
-    Returns:
-        ProcessingInput: A ``ProcessingInput`` object configured with the
-            local path. The path will be uploaded to S3 during
-            ``Processor.run()``.
+    :param input_name: The name for this processing input.
+    :param local_path: A local file or directory path to use as input.
+        This will be uploaded to S3 automatically before the
+        processing job starts.
+    :param destination: The container path where the input data will be
+        made available (e.g. ``'/opt/ml/processing/input/data'``).
+    :param s3_data_type: The S3 data type (default: ``'S3Prefix'``).
+    :param s3_input_mode: The input mode (default: ``'File'``).
+    :returns: A ``ProcessingInput`` configured with the local path.
+        The path will be uploaded to S3 during ``Processor.run()``.
+    :raises ValueError: If ``input_name``, ``local_path``, or ``destination`` is empty.
 
-    Raises:
-        ValueError: If ``input_name`` or ``local_path`` is empty.
+    Example::
 
-    Example:
         >>> inp = processing_input_from_local(
         ...     input_name="my-data",
         ...     local_path="/home/user/data/",
@@ -135,6 +130,10 @@ def processing_input_from_local(
         raise ValueError(
             f"local_path must be a non-empty string, got: {local_path!r}"
         )
+    if not destination:
+        raise ValueError(
+            f"destination must be a non-empty string, got: {destination!r}"
+        )
     return ProcessingInput(
         input_name=input_name,
         s3_input=ProcessingS3Input(
@@ -149,38 +148,33 @@ def processing_input_from_local(
 def create_processing_input(
     source: str,
     destination: str,
-    input_name: str,
-    s3_data_type: str = "S3Prefix",
-    s3_input_mode: str = "File",
+    input_name: str | None = None,
+    s3_data_type: Literal["S3Prefix", "ManifestFile"] = "S3Prefix",
+    s3_input_mode: Literal["File", "Pipe"] = "File",
 ) -> ProcessingInput:
-    """Creates a ProcessingInput from a local path or S3 URI.
+    """Create a ProcessingInput from a local path or S3 URI.
 
     This factory provides V2-like ergonomics where users pass a ``source``
     parameter that can be either a local file/directory path or an S3 URI.
     If ``source`` is a local path (does not start with ``s3://``), it will
     be automatically uploaded to S3 when the processor runs.
 
-    Args:
-        source: A local file/directory path or S3 URI. Local paths will
-            be uploaded to S3 automatically before the processing job
-            starts.
-        destination: The container path where the input data will be
-            made available (e.g. ``/opt/ml/processing/input/data``).
-        input_name: The name for this processing input.
-        s3_data_type: The S3 data type. Valid values: ``'S3Prefix'``,
-            ``'ManifestFile'`` (default: ``'S3Prefix'``).
-        s3_input_mode: The input mode. Valid values: ``'File'``,
-            ``'Pipe'`` (default: ``'File'``).
-
-    Returns:
-        ProcessingInput: A ``ProcessingInput`` object. If ``source`` is a
-            local path, it is stored in ``ProcessingS3Input.s3_uri`` and
-            will be uploaded to S3 during ``Processor.run()``.
-
-    Raises:
-        ValueError: If ``source``, ``destination``, or ``input_name`` is empty.
+    :param source: A local file/directory path or S3 URI. Local paths will
+        be uploaded to S3 automatically before the processing job starts.
+    :param destination: The container path where the input data will be
+        made available (e.g. ``'/opt/ml/processing/input/data'``).
+    :param input_name: The name for this processing input. If ``None``,
+        a name will be auto-generated by ``Processor._normalize_inputs()``
+        (default: ``None``).
+    :param s3_data_type: The S3 data type (default: ``'S3Prefix'``).
+    :param s3_input_mode: The input mode (default: ``'File'``).
+    :returns: A ``ProcessingInput`` object. If ``source`` is a local path,
+        it is stored in ``ProcessingS3Input.s3_uri`` and will be uploaded
+        to S3 during ``Processor.run()``.
+    :raises ValueError: If ``source`` or ``destination`` is empty.
+
+    Example::
 
-    Example:
         >>> # Using a local path
         >>> inp = create_processing_input(
         ...     source="/home/user/data/",
@@ -201,10 +195,6 @@ def create_processing_input(
         raise ValueError(
             f"destination must be a non-empty string, got: {destination!r}"
         )
-    if not input_name:
-        raise ValueError(
-            f"input_name must be a non-empty string, got: {input_name!r}"
-        )
     return ProcessingInput(
         input_name=input_name,
         s3_input=ProcessingS3Input(
@@ -561,22 +551,6 @@ def _normalize_inputs(self, inputs=None, kms_key=None):
                 if file_input.input_name is None:
                     file_input.input_name = f"input-{count}"
 
-                # Support ad-hoc 'source' attribute for V2-like ergonomics.
-                # If a user sets file_input.source (e.g. via monkey-patching or
-                # a subclass), populate s3_input.s3_uri from it so the existing
-                # upload logic handles it.
-                _source = getattr(file_input, "source", None)
-                if _source is not None:
-                    if file_input.s3_input is None:
-                        file_input.s3_input = ProcessingS3Input(
-                            s3_uri=_source,
-                            local_path="/opt/ml/processing/input",
-                            s3_data_type="S3Prefix",
-                            s3_input_mode="File",
-                        )
-                    else:
-                        file_input.s3_input.s3_uri = _source
-
                 if file_input.dataset_definition:
                     normalized_inputs.append(file_input)
                     continue

sagemaker-core/tests/unit/test_processing.py

@@ -527,24 +527,36 @@ def test_processing_input_from_local_with_custom_data_type_and_mode(self):
         assert result.s3_input.s3_data_type == "ManifestFile"
         assert result.s3_input.s3_input_mode == "Pipe"
 
-    def test_processing_input_from_local_empty_input_name_raises(self):
-        """processing_input_from_local should raise ValueError for empty input_name."""
+    @pytest.mark.parametrize("input_name", ["", None])
+    def test_processing_input_from_local_invalid_input_name_raises(self, input_name):
+        """processing_input_from_local should raise ValueError for empty or None input_name."""
         with pytest.raises(ValueError, match="input_name must be a non-empty string"):
             processing_input_from_local(
-                input_name="",
+                input_name=input_name,
                 local_path="/tmp/data",
                 destination="/opt/ml/processing/input",
             )
 
-    def test_processing_input_from_local_empty_local_path_raises(self):
-        """processing_input_from_local should raise ValueError for empty local_path."""
+    @pytest.mark.parametrize("local_path", ["", None])
+    def test_processing_input_from_local_invalid_local_path_raises(self, local_path):
+        """processing_input_from_local should raise ValueError for empty or None local_path."""
         with pytest.raises(ValueError, match="local_path must be a non-empty string"):
             processing_input_from_local(
                 input_name="data",
-                local_path="",
+                local_path=local_path,
                 destination="/opt/ml/processing/input",
             )
 
+    @pytest.mark.parametrize("destination", ["", None])
+    def test_processing_input_from_local_invalid_destination_raises(self, destination):
+        """processing_input_from_local should raise ValueError for empty or None destination."""
+        with pytest.raises(ValueError, match="destination must be a non-empty string"):
+            processing_input_from_local(
+                input_name="data",
+                local_path="/tmp/data",
+                destination=destination,
+            )
+
     def test_processing_input_from_local_with_pipeline_config_uses_pipeline_s3_path(
         self, mock_session
     ):
@@ -574,6 +586,9 @@ def test_processing_input_from_local_with_pipeline_config_uses_pipeline_s3_path(
             ) as mock_upload:
                 result = processor._normalize_inputs([inp])
                 mock_upload.assert_called_once()
+                # Verify the session is passed correctly to the upload call
+                call_kwargs = mock_upload.call_args
+                assert call_kwargs[1]["sagemaker_session"] == mock_session
                 assert result[0].s3_input.s3_uri.startswith("s3://")
 
 
@@ -614,32 +629,34 @@ def test_create_processing_input_default_s3_data_type_and_input_mode(self):
         assert result.s3_input.s3_data_type == "S3Prefix"
         assert result.s3_input.s3_input_mode == "File"
 
-    def test_create_processing_input_empty_source_raises(self):
-        """create_processing_input should raise ValueError for empty source."""
+    @pytest.mark.parametrize("source", ["", None])
+    def test_create_processing_input_invalid_source_raises(self, source):
+        """create_processing_input should raise ValueError for empty or None source."""
         with pytest.raises(ValueError, match="source must be a non-empty string"):
             create_processing_input(
-                source="",
+                source=source,
                 destination="/opt/ml/processing/input",
                 input_name="data",
             )
 
-    def test_create_processing_input_empty_destination_raises(self):
-        """create_processing_input should raise ValueError for empty destination."""
+    @pytest.mark.parametrize("destination", ["", None])
+    def test_create_processing_input_invalid_destination_raises(self, destination):
+        """create_processing_input should raise ValueError for empty or None destination."""
         with pytest.raises(ValueError, match="destination must be a non-empty string"):
             create_processing_input(
                 source="/tmp/data",
-                destination="",
+                destination=destination,
                 input_name="data",
             )
 
-    def test_create_processing_input_empty_input_name_raises(self):
-        """create_processing_input should raise ValueError for empty input_name."""
-        with pytest.raises(ValueError, match="input_name must be a non-empty string"):
-            create_processing_input(
-                source="/tmp/data",
-                destination="/opt/ml/processing/input",
-                input_name="",
-            )
+    def test_create_processing_input_with_none_input_name_succeeds(self):
+        """create_processing_input should succeed with input_name=None (auto-generated)."""
+        result = create_processing_input(
+            source="/tmp/data",
+            destination="/opt/ml/processing/input",
+        )
+        assert result.input_name is None
+        assert result.s3_input.s3_uri == "/tmp/data"
 
 
 class TestNormalizeInputsLocalPathUpload: