fix: correct handling of exceptions in input functions that are gener…

…ators (#1536)

* fix: correct handling of exceptions in input functions that are generators

* fmt

snakemake/exceptions.py

@@ -510,7 +510,7 @@ class CheckSumMismatchException(WorkflowError):
 class IncompleteCheckpointException(Exception):
     def __init__(self, rule, targetfile):
         super().__init__(
-            "The requested checkpoint output is not yet created."
+            "The requested checkpoint output is not yet created. "
             "If you see this error, you have likely tried to use "
             "checkpoint output outside of an input function, or "
             "you have tried to call an input function directly "

snakemake/executors/__init__.py

@@ -465,6 +465,9 @@ def get_python_executable(self):
     def get_envvar_declarations(self):
         return ""
 
+    def get_job_args(self, job, **kwargs):
+        return f"{super().get_job_args(job, **kwargs)} --quiet"
+
     def run(self, job, callback=None, submit_callback=None, error_callback=None):
         super()._run(job)
 
@@ -581,7 +584,6 @@ def run_group_job(self, job):
 
     def spawn_job(self, job):
         cmd = self.format_job_exec(job)
-        print(cmd)
         try:
             subprocess.check_call(cmd, shell=True)
         except subprocess.CalledProcessError as e:

snakemake/rules.py

@@ -5,6 +5,7 @@
 
 import os
 import re
+import types
 from snakemake.path_modifier import PATH_MODIFIER_FLAG
 import sys
 import inspect
@@ -753,6 +754,11 @@ def apply_input_function(
 
         try:
             value = func(Wildcards(fromdict=wildcards), **_aux_params)
+            if isinstance(value, types.GeneratorType):
+                # generators should be immediately collected here,
+                # otherwise we would miss any exceptions and
+                # would have to capture them again later.
+                value = list(value)
         except IncompleteCheckpointException as e:
             value = incomplete_checkpoint_func(e)
             incomplete = True

tests/test_github_issue261/Snakefile

@@ -0,0 +1,25 @@
+import random
+
+checkpoint random:
+    output:
+        target = 'test1/{target}/{config}.txt'
+
+    run:
+        with open(output.target, 'w') as ftg:
+            for i in range(2, 10):
+                ftg.write(f'test1/{wildcards.target}/{wildcards.config}/v{i}\n')
+
+rule process:
+    output:
+        touch('test1/{target}/{config}/v{index}')
+
+def genetate_inputs(wildcards):
+    with checkpoints.random.get(**wildcards).output[0].open() as fran:
+        for line in fran.readlines():
+            yield line.rstrip()
+
+rule generate:
+    input:
+        genetate_inputs
+    output:
+        touch('test1/{target}/{config}.done')

tests/tests.py

@@ -1567,6 +1567,11 @@ def test_incomplete_params():
     run(dpath("test_incomplete_params"), dryrun=True, printshellcmds=True)
 
 
+@skip_on_windows
+def test_github_issue261():
+    run(dpath("test_github_issue261"), targets=["test1/target1/config1.done"])
+
+
 @skip_on_windows  # no pipe support on windows
 def test_pipe_depend():
     run(dpath("test_pipe_depend"), shouldfail=True)