Create your own tasks and recipe

A task is an ordinary python object, with an __init__ method and a __call__ method, which has to subclass goatlib.Task.

Each task must declare a category in order to be executed, let’s chose the “extract” category.

The category must be declared as a class attribute because Scheduler will manage Tasks subclasses and he must have access to their categories without having to instantiate them.

import tarfile

class TarExtract(Task):
    category = "extract"

    def __call__(self, archive_path, destination_path):
        tar_object = tarfile(archive_path)
        tar_object.extractall(destination_path)

This code is very simple, but we can see that it needs two arguments, one for the archive path and one for destination path. If this case, task must declare them as inputs in our task. This will make a difference with the last code, task will not receive archive paths and destination paths, instead it will receive two Items representing an archive and an directory. Then it will be able to retrieve their paths from these objects.

class TarExtract(Task):
    ...
    inputs = {"archive" : "file.archive", "destination" : "directory"}
    ...

One more thing, any task must return informations about it’s status, if current code works, task should have the status SUCCESS.

from pyti.goatlib import Task, SUCCESS

class TarExtract(Task):
    ...
    def __call__(self, archive, destination):
        ...
        return SUCCESS, []

In last exemple, task return the status of the task and a list, empty for now, of Items. This task should return a new Item representing the directory where the archive has been extracted.

from pyti.goatlib import Task, Item, SUCCESS
from os.path import splitext, join
import tarfile

class TarExtract(Task):
    category = "extract"

    inputs = {"archive" : "file.archive", "destination" : "directory"}

    def __call__(self, archive, destination):
        tar_object = tarfile.open(archive['path'])
        tar_object.extractall(destination['path'])

        extracted_path = join(destination['path'],
                              splitext(archive['path'])[1]
                              )

        return SUCCESS, [Item(types = ["directory"], data = {"path" : extracted_path})]

Create a recipe