feat: allow to mark input files of rules as ancient via the API or command line interface (and thereby also via workflow specific profiles). Putting this into a workflow specific profile (or specifying as argument) allows to overrule rerun triggers caused by file modification dates where the user knows better.

johanneskoester · johanneskoester · commit 84f4e79db054 · 2024-10-24T19:47:25.000+02:00
diff --git a/snakemake/cli.py b/snakemake/cli.py
@@ -3,6 +3,7 @@
 __email__ = "johannes.koester@uni-due.de"
 __license__ = "MIT"
 
+from collections import defaultdict
 import os
 import re
 import sys
@@ -102,6 +103,32 @@ def fallback(orig_value):
     )
 
 
+def parse_consider_ancient(args):
+    errmsg = (
+        "Invalid --consider-ancient definition: entries have to be defined as "
+        "RULE=INPUTITEMS pairs, with INPUTITEMS being a list of input items of the "
+        "rule (given as name or index (0-based)), separated by commas."
+    )
+
+    def parse_item(item):
+        try:
+            return int(item)
+        except ValueError:
+            if item.isidentifier():
+                return item
+            else:
+                raise ValueError(f"{errmsg} (Unparsable value: {repr(item)})")
+
+    consider_ancient = defaultdict(set)
+
+    if args is not None:
+        for entry in args:
+            rule, items = parse_key_value_arg(entry, errmsg=errmsg, strip_quotes=True)
+            items = items.split(",")
+            consider_ancient[rule] = {parse_item(item) for item in items}
+    return consider_ancient
+
+
 def parse_set_resources(args):
     errmsg = (
         "Invalid resource definition: entries have to be defined as RULE:RESOURCE=VALUE, with "
@@ -730,6 +757,19 @@ def get_argument_parser(profiles=None):
             "output in your workflow updated."
         ),
     )
+    group_exec.add_argument(
+        "--consider-ancient",
+        metavar="RULE=INPUTITEMS",
+        nargs="+",
+        default=dict(),
+        parse_func=parse_consider_ancient,
+        help="Consider given input items of given rules as ancient, i.e. not triggering "
+        "re-runs if they are newer than the output files. "
+        "Putting this into a workflow specific profile (or specifying as argument) "
+        "allows to overrule rerun triggers caused by file modification dates where the "
+        "user knows better.",
+    )
+
     group_exec.add_argument(
         "--prioritize",
         "-P",
diff --git a/snakemake/rules.py b/snakemake/rules.py
@@ -272,10 +272,15 @@ def set_input(self, *input, **kwinput):
         Arguments
         input -- the list of input files
         """
-        for item in input:
-            self._set_inoutput_item(item)
+
+        consider_ancient = self.workflow.dag_settings.consider_ancient.get(self.name)
+
+        for i, item in enumerate(input):
+            self._set_inoutput_item(item, mark_ancient=i)
         for name, item in kwinput.items():
-            self._set_inoutput_item(item, name=name)
+            self._set_inoutput_item(
+                item, name=name, mark_ancient=name in consider_ancient
+            )
 
     @property
     def output(self):
@@ -400,7 +405,7 @@ def _update_item_wildcard_constraints(self, item):
         except ValueError as e:
             raise WorkflowError(e, snakefile=self.snakefile, lineno=self.lineno)
 
-    def _set_inoutput_item(self, item, output=False, name=None):
+    def _set_inoutput_item(self, item, output=False, name=None, mark_ancient=False):
         """
         Set an item to be input or output.
 
@@ -462,12 +467,17 @@ def _set_inoutput_item(self, item, output=False, name=None):
                             )
                         )
 
-            # add the rule to the dependencies
             if rule_dependency is not None:
+                # add the rule to the dependencies
                 self.dependencies[item] = rule_dependency
+
             if output:
                 item = self._update_item_wildcard_constraints(item)
+                if self.workflow.storage_settings.all_temp:
+                    # mark as temp if all output files shall be marked as temp
+                    item = flag(item, "temp")
             else:
+                # input
                 if (
                     contains_wildcard_constraints(item)
                     and self.workflow.exec_mode != ExecMode.SUBPROCESS
@@ -477,10 +487,8 @@ def _set_inoutput_item(self, item, output=False, name=None):
                             self
                         )
                     )
-
-            if self.workflow.storage_settings.all_temp and output:
-                # mark as temp if all output files shall be marked as temp
-                item = flag(item, "temp")
+                if mark_ancient:
+                    item = flag(item, "ancient")
 
             # record rule if this is an output file output
             _item = IOFile(item, rule=self)
@@ -520,8 +528,8 @@ def _set_inoutput_item(self, item, output=False, name=None):
         else:
             try:
                 start = len(inoutput)
-                for i in item:
-                    self._set_inoutput_item(i, output=output)
+                for subitem in item:
+                    self._set_inoutput_item(subitem, output=output)
                 if name:
                     # if the list was named, make it accessible
                     inoutput._set_name(name, start, end=len(inoutput))
diff --git a/snakemake/settings/types.py b/snakemake/settings/types.py
@@ -3,7 +3,7 @@
 import os
 from pathlib import Path
 import re
-from typing import Any, Optional
+from typing import Any, List, Optional, Union
 from typing import Mapping, Sequence, Set
 
 import immutables
@@ -199,6 +199,9 @@ class DAGSettings(SettingsBase):
     allowed_rules: AnySet[str] = frozenset()
     rerun_triggers: AnySet[RerunTrigger] = RerunTrigger.all()
     max_inventory_wait_time: int = 20
+    consider_ancient: Mapping[str, AnySet[Union[str, int]]] = field(
+        default_factory=dict
+    )
 
     def _check(self):
         if self.batch is not None and self.forceall:
