Skip to content
Permalink
Browse files
fix: fixed handling of input functions and unpack when using the pref…
…ix setting of module definitions (#1553)
  • Loading branch information
johanneskoester committed Apr 1, 2022
1 parent 6866134 commit d561e041a8919717046be0d39f197b2c6b937cb7
Show file tree
Hide file tree
Showing 8 changed files with 61 additions and 10 deletions.
@@ -5,7 +5,7 @@

import os
from snakemake.exceptions import WorkflowError
from snakemake.io import is_flagged, AnnotatedString, flag, get_flag_value
from snakemake.io import is_callable, is_flagged, AnnotatedString, flag, get_flag_value


PATH_MODIFIER_FLAG = "path_modified"
@@ -57,6 +57,7 @@ def replace_prefix(self, path, property=None):
or os.path.isabs(path)
or path.startswith("..")
or is_flagged(path, "remote_object")
or is_callable(path)
):
# no replacement
return path
@@ -78,9 +79,9 @@ def replace_prefix(self, path, property=None):
else:
# no matching prefix
return path

# prefix case
return self.prefix + path
else:
# prefix case
return self.prefix + path

def apply_default_remote(self, path):
"""Apply the defined default remote provider to the given path and return the updated _IOFile.
@@ -734,7 +734,7 @@ def apply_input_function(
incomplete_checkpoint_func=lambda e: None,
raw_exceptions=False,
groupid=None,
**aux_params
**aux_params,
):
incomplete = False
if isinstance(func, _IOFile):
@@ -811,24 +811,27 @@ def _apply_wildcards(
incomplete_checkpoint_func=incomplete_checkpoint_func,
is_unpack=is_unpack,
groupid=groupid,
**aux_params
**aux_params,
)

if is_unpack and not incomplete:
if not allow_unpack:
raise WorkflowError(
"unpack() is not allowed with params. "
"Simply return a dictionary which can be directly ."
"used, e.g. via {params[mykey]}."
"used, e.g. via {params[mykey]}.",
rule=self,
)
# Sanity checks before interpreting unpack()
if not isinstance(item, (list, dict)):
raise WorkflowError(
"Can only use unpack() on list and dict", rule=self
f"Can only use unpack() on list and dict, but {item} was returned.",
rule=self,
)
if name:
raise WorkflowError(
"Cannot combine named input file with unpack()", rule=self
f"Cannot combine named input file (name {name}) with unpack()",
rule=self,
)
# Allow streamlined code with/without unpack
if isinstance(item, list):
@@ -1067,7 +1070,7 @@ def apply(name, res, threads=None):
attempt=attempt,
incomplete_checkpoint_func=lambda e: 0,
raw_exceptions=True,
**aux
**aux,
)
except (Exception, BaseException) as e:
raise InputFunctionException(e, rule=self, wildcards=wildcards)
@@ -0,0 +1,22 @@
rule all:
input:
"results/all.done",


module other:
snakefile:
"module.smk"
prefix:
"results/other"


use rule * from other as other_*


rule all_impl:
output:
out=touch("results/all.done"),
input:
b=rules.other_b.output.b,
shell:
"cp {input} {output}"
Empty file.
@@ -0,0 +1,21 @@
rule a:
output:
a=touch("results/a.txt"),


def unpack_a(wildcards):
return {"a": rules.a.output.a}


def lambda_a(wildcards):
return rules.a.output.a


rule b:
input:
unpack(unpack_a),
# a = lambda_a,
output:
b="results/b.txt",
shell:
"cp {input.a} {output}"
@@ -1591,5 +1591,9 @@ def test_github_issue1542():
run(dpath("test_github_issue1542"), dryrun=True)


def test_github_issue1498():
run(dpath("test_github_issue1498"))


def test_cleanup_metadata_fail():
run(dpath("test09"), cleanup_metadata=["xyz"])

0 comments on commit d561e04

Please sign in to comment.