Arvados

h1. Python SDK

(design draft)

<pre><code class="python">

#!/usr/bin/env python

from arvados import CrunchJob

import examplelib

import re

class NormalizeMatchingFiles(CrunchJob):

    @CrunchJob.task()

    def grep_files(self):

        # CrunchJob instantiates input parameters based on the

        # dataclass attribute.  When we ask for the input parameter,

        # CrunchJob sees that it's a Collection, and returns a

        # CollectionReader object.

        for filename in self.job_param('input').filenames():

            self.grep_file(filename)

    @CrunchJob.task()

    def grep_file(self, filename):

        regexp = re.compile(self.job_param('pattern'))

        with self.job_param('input').open(filename) as in_file:

            for line in in_file:

                if regexp.search(line):

                    self.normalize(filename)

                    break

    # examplelib is already multi-threaded and will peg the whole

    # compute node.  These tasks should run sequentially.

    @CrunchJob.task(parallel_with=[])

    def normalize(self, filename):

        output = examplelib.frob(self.job_param('input').mount_path(filename))

        # self.output is a CollectionWriter.  When this task method finishes,

        # CrunchJob checks if we wrote anything to it.  If so, it takes care

        # of finishing the upload process, and sets this task's output to the

        # Collection UUID.

        with self.output.open(filename) as out_file:

            out_file.write(output)

if __name__ == '__main__':

    NormalizeMatchingFiles(task0='grep_files').main()

</code></pre>