[API] Making source parameter for ingest API optional

mlrun/api/api/endpoints/feature_sets.py

@@ -158,8 +158,9 @@ def _has_v3io_path(data_source, data_targets, feature_set):
             # If the target does not have a path (i.e. default target), then retrieve the default path from config.
             paths.append(target.path or get_default_prefix_for_target(target.kind))
 
-    if data_source:
-        paths.append(data_source.path)
+    source = data_source or feature_set.spec.source
+    if source:
+        paths.append(source.path)
 
     return any(
         path and (path.startswith("v3io://") or path.startswith("v3ios://"))
@@ -188,8 +189,9 @@ def ingest_feature_set(
     # Need to override the default rundb since we're in the server.
     feature_set._override_run_db(db_session)
 
-    data_source = DataSource.from_dict(ingest_parameters.source.dict())
-    data_targets = None
+    data_source = data_targets = None
+    if ingest_parameters.source:
+        data_source = DataSource.from_dict(ingest_parameters.source.dict())
     if ingest_parameters.targets:
         data_targets = [
             DataTargetBase.from_dict(data_target.dict())

mlrun/api/schemas/feature_store.py

@@ -134,7 +134,7 @@ class Config:
 
 
 class FeatureSetIngestInput(BaseModel):
-    source: DataSource
+    source: Optional[DataSource]
     targets: Optional[List[DataTarget]]
     infer_options: Optional[int]
 

mlrun/feature_store/api.py

@@ -180,15 +180,21 @@ def ingest(
     :param mlrun_context: mlrun context (when running as a job), for internal use !
     :param spark_context: local spark session for spark ingestion, example for creating the spark context:
                           `spark = SparkSession.builder.appName("Spark function").getOrCreate()`
-    :param start_time     datetime/string, low limit of time needed to be filtered. format '2020-11-01 17:33:15'
-    :param end_time       datetime/string, high limit of time needed to be filtered. format '2020-12-01 17:33:15'
+    :param start_time:    datetime/string, low limit of time needed to be filtered. format '2020-11-01 17:33:15'
+    :param end_time:      datetime/string, high limit of time needed to be filtered. format '2020-12-01 17:33:15'
     """
+    if featureset:
+        if isinstance(featureset, str):
+            featureset = get_feature_set_by_uri(featureset)
+        # feature-set spec always has a source property that is not None. It may be default-constructed, in which
+        # case the path will be 'None'. That's why we need a special check
+        if source is None and featureset.has_valid_source():
+            source = featureset.spec.source
+
     if not mlrun_context and (not featureset or source is None):
         raise mlrun.errors.MLRunInvalidArgumentError(
             "feature set and source must be specified"
         )
-    if featureset and isinstance(featureset, str):
-        featureset = get_feature_set_by_uri(featureset)
 
     if run_config:
         # remote job execution

mlrun/feature_store/feature_set.py

@@ -319,6 +319,11 @@ def set_targets(self, targets=None, with_defaults=True, default_final_state=None
         if default_final_state:
             self.spec.graph.final_state = default_final_state
 
+    def has_valid_source(self):
+        """check if object's spec has a valid (non empty) source definition"""
+        source = self.spec.source
+        return source is not None and source.path is not None and source.path != "None"
+
     def add_entity(self, entity, name=None):
         """add/set an entity"""
         self._spec.entities.update(entity, name)