Snakemake循环生成规则失效：为何仅生成最后一个规则的输出？

问题重现 #

尝试通过遍历列表批量生成Snakemake规则，代码如下：

l=[
  {"name": "one", "output": "1.txt", "shell": "touch 1.txt"},
  {"name": "two", "output": "2.txt", "shell": "touch 2.txt"}
]
for i in l:
  rule:
    name: i["name"]
    output: i["output"]
    shell: i["shell"]


rule all:
  input:
    "1.txt",
    "2.txt"

执行时任务统计显示包含all、one、two三个任务：

Building DAG of jobs...
Using shell: /bin/bash
Provided cores: 10
Rules claiming more threads will be scaled down.
Job stats:
job      count    min threads    max threads
-----  -------  -------------  -------------
all          1              1              1
one          1              1              1
two          1              1              1
total        3              1              1

但实际执行抛出错误：

rule two:
    output: 2.txt
    jobid: 2
    reason: Forced execution
    resources: tmpdir=/var/folders/7j/s10tsc1d0qv1rvl0_k76yzkm0000gn/T

Waiting at most 5 seconds for missing files.
MissingOutputException in rule one in file /Users/grasseld/Documents/git_repos/OCEAN/GLOPO/glopo-cli/programs/workflow.smk, line 6:
Job 1  completed successfully, but some output files are missing. Missing files after 5 seconds. This might be due to filesystem latency. If that is the case, consider to increase the wait time with --latency-wait:
1.txt

手动编写两个规则则可正常运行：

rule one:
  output: "1.txt"
  shell: "touch 1.txt"

rule two:
  output: "2.txt"
  shell: "touch 2.txt"


rule all:
  input:
    "1.txt",
    "2.txt"  

原因分析 #

问题出在Python的变量作用域与延迟绑定：

• 循环中定义规则时，i是指向列表元素的引用，而非值的拷贝。
• Snakemake解析Snakefile时仅收集规则框架，但规则的output和shell参数是延迟求值的——直到执行规则时才会读取i的当前值。
• 循环结束后，i已经指向列表最后一个元素（{"name": "two", ...}），因此所有循环生成的规则最终都使用了最后一个元素的参数：
  • 名为one的规则实际执行的shell命令是touch 2.txt，不会生成1.txt，直接触发MissingOutputException。

解决方案 #

要避免延迟绑定问题，需在循环中捕获当前迭代的i的具体值，常用两种方式：

方式1：使用exec动态生成规则代码 #

通过字符串拼接生成每个规则的定义，再用exec执行，确保每个规则的参数都是当前迭代的固定值：

l=[
  {"name": "one", "output": "1.txt", "shell": "touch 1.txt"},
  {"name": "two", "output": "2.txt", "shell": "touch 2.txt"}
]

for i in l:
    rule_code = f'''
rule {i["name"]}:
    output: "{i["output"]}"
    shell: "{i["shell"]}"
    '''
    exec(rule_code)

rule all:
  input:
    "1.txt",
    "2.txt"

方式2：利用闭包捕获当前值 #

通过嵌套函数捕获当前迭代的i，强制参数在规则定义时就绑定到当前值：

l=[
  {"name": "one", "output": "1.txt", "shell": "touch 1.txt"},
  {"name": "two", "output": "2.txt", "shell": "touch 2.txt"}
]

def make_rule(rule_info):
    rule:
        name: rule_info["name"]
        output: rule_info["output"]
        shell: rule_info["shell"]

for i in l:
    make_rule(i)

rule all:
  input:
    "1.txt",
    "2.txt"

内容的提问来源于stack exchange，提问作者dgroche