Work in progress: Foreach to iterate over files in a directory to support incremental processing

dvc/parsing/__init__.py

@@ -1,8 +1,10 @@
 import logging
 import os
+import re
 from collections.abc import Mapping, Sequence
 from copy import deepcopy
 from itertools import product
+from pathlib import Path
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -158,6 +160,18 @@ def __init__(self, repo: "Repo", wdir: str, d: dict):
         except ContextError as exc:
             format_and_raise(exc, "'vars'", self.relpath)
 
+        stages = d["stages"]
+        for stage in stages:
+            if "foreach" in stages[stage] and "dep_files" in stages[stage]:
+                varname = re.match(r"\${(\w+)}", stages[stage]["foreach"]).group(1)
+                # Hardcoding prefixes to be from a given directory
+                directory = Path(stages[stage]["dep_files"])
+                assert directory.is_dir()
+                self.context[varname] = [
+                    {"name": path.name, "stem": path.stem}
+                    for path in Path(directory).glob("*")
+                ]
+
         # we use `tracked_vars` to keep a dictionary of used variables
         # by the interpolated entries.
         self.tracked_vars: Dict[str, Mapping] = {}

dvc/schema.py

@@ -117,6 +117,7 @@ def validator(data):
 FOREACH_IN = {
     Required(FOREACH_KWD): Any(dict, list, str),
     Required(DO_KWD): STAGE_DEFINITION,
+    "dep_files": str,
 }
 SINGLE_PIPELINE_STAGE_SCHEMA = {
     str: either_or(STAGE_DEFINITION, FOREACH_IN, [FOREACH_KWD, DO_KWD])

dvc/stage/params.py

@@ -12,3 +12,4 @@ class StageParams:
     PARAM_METRICS = "metrics"
     PARAM_PLOTS = "plots"
     PARAM_DESC = "desc"
+    PARAM_INCDIR = "dep_files"