# New Object orientated syntax for Ruffus in Version 2.6

> Ruffus Pipelines can now be created and manipulated directly using Pipeline and Task objects instead of via decorators.

Note

You may want to go through the worked_example first.

## Syntax

> This traditional Ruffus code:

> >     from ruffus import *

> >     # task function
> >     starting_files = ["input/a.fasta","input/b.fasta"]
> >     @transform(input      = starting_files,
> >                filter     = suffix('.fasta'),
> >                output     = '.sam',
> >                output_dir = "output")
> >     def map_dna_sequence(input_file, output_file) :
> >         pass

> >     pipeline_run()

> Can also be written as:

> >     from ruffus import *

> >     # undecorated task function
> >     def map_dna_sequence(input_file, output_file) :
> >         pass

> >     starting_files = ["input/a.fasta","input/b.fasta"]

> >     #   make ruffus Pipeline() object
> >     my_pipeline = Pipeline(name = "test")
> >     my_pipeline.transform(task_func  = map_dna_sequence,
> >                           input      = starting_files,
> >                           filter     = suffix('.fasta'),
> >                           output     = '.sam',
> >                           output_dir = "output")

> >     my_pipeline.run()

> The two different syntax are almost identical:
> The first parameter task_func=`your_python_function` is mandatory.
> Otherwise, all other parameters are in the same order as before, and can be given by position or as named arguments.

## Advantages

> These are some of the advantages of the new syntax:

> 1. Pipeline topology is assembled in one place

>    This is a matter of personal preference.

>    Nevertheless, using decorators to locally annotate python functions with pipeline parameters arguably helps separation of concerns.

> 2. Pipelines can be created on the fly

>    For example, using parameters parsed from configuration files.

>    Ruffus pipelines no longer have to be defined at global scope.

> 3. Reuse common sub-pipelines

>    Shared sub pipelines can be created from discrete python modules and joined together as needed. Bioinformaticists may have “mapping”, “aligning”, “variant-calling” sub-pipelines etc.

> 4. Multiple Tasks can share the same python function

>    Tasks are normally referred to by their associated functions (as with decoratored Ruffus tasks). However, you can also disambiguate Tasks by specifying their name directly.

> 5. Pipeline topology can be specified at run time

>    Some (especially bioinformatics) tasks require binary merging. This can be very inconvenient.

>    For example, if we have 8 data files, we need three successive rounds of merging (8->4->2->1) or three tasks) to produce the output. But if we are given 10 data files, we now find that we needed to have four tasks for four rounds of merging (10->5->3->2->1).

>    There was previously no easy way to arrange different Ruffus topologies in response to the data. Now we can add as many extra merging tasks to our pipeline (all sharing the same underlying python function) as needed.

## Compatability

> - The changes are fully backwards compatibile. All valid Ruffus code continues to work
> - Decorators and `Pipeline` objects can be used interchangeably:
> Decorated functions are automatically part of a default constructed `Pipeline` named `"main"`.

>       main_pipeline = Pipeline.pipelines["main"]

> In the following example, a pipeline using the Ruffus with classes syntax (1) and (3) has a traditionally decorated task function in the middle (2).

> >     from ruffus import *

> >     # get default pipeline
> >     main_pipeline = Pipeline.pipelines["main"]

> >     # undecorated task functions
> >     def compress_sam_to_bam(input_file, output_file) :
> >         open(output_file, "w").close()

> >     def create_files(output_file) :
> >         open(output_file, "w").close()

> >     #
> >     #   1. Ruffus with classes
> >     #
> >     starting_files = main_pipeline.originate(create_files, ["input/a.fasta","input/b.fasta"])\
> >         .follows(mkdir("input", "output"))

> >     #
> >     #   2. Ruffus with python decorations
> >     #
> >     @transform(starting_files,
> >                suffix('.fasta'),
> >                '.sam',
> >                output_dir = "output")
> >     def map_dna_sequence(input_file, output_file) :
> >         open(output_file, "w").close()

> >     #
> >     #   3. Ruffus with classes
> >     #
> >     main_pipeline.transform(task_func   = compress_sam_to_bam,
> >                             input       = map_dna_sequence,
> >                             filter      = suffix(".sam"),
> >                             output      = ".bam")

> >     # main_pipeline.run()
> >     #    or
> >     pipeline_run()

## Class methods

> The ruffus.Pipeline class has the following self-explanatory methods:

> >     Pipeline.run(...)
> >     Pipeline.printout(...)
> >     Pipeline.printout_graph(...)

> These methods return a ruffus.Task object

> >     Pipeline.originate(...)
> >     Pipeline.transform(...)
> >     Pipeline.split(...)
> >     Pipeline.merge(...)
> >     Pipeline.mkdir(...)

> >     Pipeline.collate(...)
> >     Pipeline.subdivide(...)

> >     Pipeline.combinations(...)
> >     Pipeline.combinations_with_replacement(...)
> >     Pipeline.product(...)
> >     Pipeline.permutations(...)

> >     Pipeline.follows(...)
> >     Pipeline.check_if_uptodate(...)
> >     Pipeline.graphviz(...)

> >     Pipeline.files(...)
> >     Pipeline.parallel(...)

> A Ruffus Task can be modified with the following methods

> >     Task.active_if(...)
> >     Task.check_if_uptodate(...)
> >     Task.follows(...)
> >     Task.graphviz(...)
> >     Task.jobs_limit(...)
> >     Task.mkdir(...)
> >     Task.posttask(...)

## Call chaining

> The syntax is designed to allow call chaining:

> >     Pipeline.transform(...)\
> >         .mkdir(follows(...))\
> >         .active_if(...)\
> >         .graphviz(...)

## Referring to Tasks

> Ruffus pipelines are chained together or specified by referring to each stage or Task.

> (1) and (2) are ways to referring to tasks that Ruffus has always supported.

> (3) - (6) are new to Ruffus v 2.6 but apply to both using decorators or the new Ruffus with classes syntax.

### 1) Python function

>     @transform(prev_task, ...)
>     def next_task():
>         pass

>     pipeline.transform(input = next_task, ...)

### 2) Python function name (using output_from)

>     pipeline.transform(input = output_from("prev_task"), ...)

Note

The above (1) and (2) only work if the Python function specifies the task unambiguously in a pipeline. If you reuse the same Python function for multiple tasks, use the following methods.

Ruffus will complain with Exceptions if your code is ambiguous.

### 3) Task object

>      prev_task = pipeline.transform(...)

>      # prev_task is a Task object
>      next_task = pipeline.transform(input = prev_task, ....)

### 4) Task name (using output_from)

>     # name this task "prev_task"
>     pipeline.transform(name = "prev_task",...)

>     pipeline.transform(input = output_from("prev_task"), ....)

> Note

> Tasks from other pipelines can be referred to using full qualified names in the pipeline::task format

>     pipeline.transform(input = output_from("other_pipeline::prev_task"), ....)

### 5) Pipeline

> When we are assembling our pipeline from sub-pipelines (especially those in other modules which other people might have written) it is inconvenient to break encapsulation to find out the component Task of the subpipeline.

> In which case, the sub-pipeline author can assign particular tasks to the head and tail of the pipeline. The pipeline will be an alias for these:

>     # Note: these functions take lists
>     sub_pipeline.set_head_tasks([first_task])
>     sub_pipeline.set_tail_tasks([last_task])

>     # first_task.set_input(...)
>     sub_pipeline.set_input(input = "*.txt")

>     # (input = last_task,...)
>     main_pipeline.transform(input = sub_pipeline, ....)

> If you don’t have access to a pipeline object, you can look it up via the Pipeline object

> >     # This is the default "main" pipeline which holds decorated task functions
> >     main_pipeline = Pipeline.pipelines["main"]

> >     my_pipeline = Pipeline("test")

> >     alias_to_my_pipeline = Pipeline.pipelines["test"]

### 6) Lookup Task via the Pipeline

> We can ask a Pipeline to lookup task names, functions and function names for us.

>     # Lookup task name
>     pipeline.transform(input = pipeline["prev_task"], ....)

>     # Lookup via python function
>     pipeline.transform(input = pipeline[python_function], ....)

>     # Lookup via python function name
>     pipeline.transform(input = pipeline["python_function_name"], ....)

> This is straightforward if the lookup is unambiguous for the pipeline.

> If the names are not found in the pipeline, Ruffus will look across all pipelines.

> Any ambiguities will result in an immediate error.

> In extremis, you can use pipeline qualified names

>     # Pipeline qualified task name
>     pipeline.transform(input = pipeline["other_pipeline::prev_task"], ....)

Note

All this will be much clearer going through the worked_example.