Refactor the logic for source_metadata to be better readable. (#35370)

* Refactor source_metadata validation

* Reorg the logic for source_metadata to be more readable.

* update

* minor

* Minor

---------

Co-authored-by: Anya Li <[email protected]>

Author: anyalee0221

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_feature_set/source_metadata.py

@@ -25,60 +25,45 @@ def __init__(
         dict: Optional[Dict] = None,
         **kwargs: Any,
     ):
-        if type != "featureset":
-            if not timestamp_column:
-                msg = f"You need to provide timestamp_column for {type} feature source."
-                raise ValidationException(
-                    message=msg,
-                    no_personal_data_message=msg,
-                    error_type=ValidationErrorType.INVALID_VALUE,
-                    target=ErrorTarget.FEATURE_SET,
-                    error_category=ErrorCategory.USER_ERROR,
-                )
-
-        if type != "custom":
-            if type == "featureset":
-                if not path:
-                    msg = f"You need to provide path for {type} feature source."
-                    raise ValidationException(
-                        message=msg,
-                        no_personal_data_message=msg,
-                        error_type=ValidationErrorType.INVALID_VALUE,
-                        target=ErrorTarget.FEATURE_SET,
-                        error_category=ErrorCategory.USER_ERROR,
-                    )
-                if timestamp_column or source_delay or source_process_code:
-                    msg = f"Cannot provide timestamp_column/source_delay/source_process_code for {type} feature source."
-                    raise ValidationException(
-                        message=msg,
-                        no_personal_data_message=msg,
-                        error_type=ValidationErrorType.INVALID_VALUE,
-                        target=ErrorTarget.FEATURE_SET,
-                        error_category=ErrorCategory.USER_ERROR,
-                    )
-
-            if not (path and not dict and not source_process_code):
-                msg = f"Cannot provide source_process_code or kwargs for {type} feature source."
-                raise ValidationException(
-                    message=msg,
-                    no_personal_data_message=msg,
-                    error_type=ValidationErrorType.INVALID_VALUE,
-                    target=ErrorTarget.FEATURE_SET,
-                    error_category=ErrorCategory.USER_ERROR,
-                )
+        if type == "custom":
+            # For custom feature source
+            # Required: timestamp_column, dict and source_process_code.
+            # Not support: path.
+            if path:
+                self.throw_exception("path", type, should_provide=False)
+            if not (timestamp_column and dict and source_process_code):
+                self.throw_exception("timestamp_column/dict/source_process_code", type, should_provide=True)
+        elif type == "featureset":
+            # For featureset feature source
+            # Required: path.
+            # Not support: timestamp_column, source_delay and source_process_code.
+            if timestamp_column or source_delay or source_process_code:
+                self.throw_exception("timestamp_column/source_delay/source_process_code", type, should_provide=False)
+            if not path:
+                self.throw_exception("path", type, should_provide=True)
         else:
-            if not (dict and source_process_code and not path):
-                msg = "You cannot provide path for custom feature source."
-                raise ValidationException(
-                    message=msg,
-                    no_personal_data_message=msg,
-                    error_type=ValidationErrorType.INVALID_VALUE,
-                    target=ErrorTarget.FEATURE_SET,
-                    error_category=ErrorCategory.USER_ERROR,
-                )
+            # For other type feature source
+            # Required: timestamp_column, path.
+            # Not support: source_process_code, dict
+            if dict or source_process_code:
+                self.throw_exception("dict/source_process_code", type, should_provide=False)
+            if not (timestamp_column and path):
+                self.throw_exception("timestamp_column/path", type, should_provide=True)
         self.type = type
         self.path = path
         self.timestamp_column = timestamp_column
         self.source_delay = source_delay
         self.source_process_code = source_process_code
         self.kwargs = dict
+
+    @staticmethod
+    def throw_exception(property_names: str, type: str, should_provide: bool):
+        should_or_not = "need to" if should_provide else "cannot"
+        msg = f"You {should_or_not} provide {property_names} for {type} feature source."
+        raise ValidationException(
+            message=msg,
+            no_personal_data_message=msg,
+            error_type=ValidationErrorType.INVALID_VALUE,
+            target=ErrorTarget.FEATURE_SET,
+            error_category=ErrorCategory.USER_ERROR,
+        )