Use your own tasks

nuka’s tasks are just python class that inherit from nuka.task.Task.

Here is a simple example:

# -*- coding: utf-8 -*-
import codecs
from nuka.task import Task


class timezone(Task):

    def __init__(self, tz=None, **kwargs):
        # ensure we have a name to get a better repr() in logs
        kwargs.setdefault('name', tz)
        super(timezone, self).__init__(tz=tz, **kwargs)

    def do(self):
        """do the job: change the timezone file if needed"""
        tz = self.args['tz']
        changed = False
        with codecs.open('/etc/timezone', 'r', 'utf8') as fd:
            current_tz = fd.read().strip()
        if current_tz != tz:
            changed = True
            with codecs.open('/etc/timezone', 'w', 'utf8') as fd:
                current_tz = fd.write(tz + '\n')
        # we must return a dictionary with at least a return code and
        # the change state
        return dict(rc=0, changed=changed)

    def diff(self):
        """generate diff between actual state and task value.
        Implementing this method is not required but recommended"""
        tz = self.args['tz']
        with codecs.open('/etc/timezone', 'r', 'utf8') as fd:
            current_tz = fd.read().strip()
        diff = ''
        if current_tz != tz:
            diff = self.text_diff(current_tz, tz)
        return dict(diff=diff)

You must be sure that your code is compatible with the python binaries you use locally and remotely (2.x vs 3.x).

nuka’s builtin tasks support python 2.7 and 3.4+

As a good practice your task should be isolated in a tasks package and must only use python’s stdlib.

Once it’s done, you can use it:

# -*- coding: utf-8 -*-
import nuka
from nuka.hosts import DockerContainer

from tasks.timezone import timezone

host = DockerContainer(hostname='debian_jessie')


async def change_timezone(host):
    await timezone(tz='Europe/Paris')


nuka.run(change_timezone(host))