Source code for snakemake.checkpoints

from snakemake.exceptions import IncompleteCheckpointException, WorkflowError
from snakemake.io import checkpoint_target


[docs]class Checkpoints: """ A namespace for checkpoints so that they can be accessed via dot notation. """ def __init__(self): self.future_output = None
[docs] def register(self, rule): setattr(self, rule.name, Checkpoint(rule, self))
[docs]class Checkpoint: __slots__ = ["rule", "checkpoints"] def __init__(self, rule, checkpoints): self.rule = rule self.checkpoints = checkpoints
[docs] def get(self, **wildcards): missing = self.rule.wildcard_names.difference(wildcards.keys()) if missing: raise WorkflowError( "Missing wildcard values for {}".format(", ".join(missing)) ) output, _ = self.rule.expand_output(wildcards) if self.checkpoints.future_output is None or any( (not f.exists or f in self.checkpoints.future_output) for f in output ): raise IncompleteCheckpointException(self.rule, checkpoint_target(output[0])) return CheckpointJob(self.rule, output)
[docs]class CheckpointJob: __slots__ = ["rule", "output"] def __init__(self, rule, output): self.output = output self.rule = rule