



  

pytask documentation

 [pytask](index.html#document-index) 

 latest

 

Using pytask

* [Starting Workers](index.html#document-starting_workers)

* [Writing Tasks](index.html#document-writing_tasks)

* [pytask API](index.html#document-api)

pytask Clusters

* [Clustering pytask](index.html#document-clustering_pytask)

* [Fault Tolerance](index.html#document-fault_tolerance)

Elsewhere

* [pytask on GitHub](http://github.com/Oxygem/pytask)

[pytask](index.html#document-index)

* [Docs](index.html#document-index) »

* pytask documentation

* [Edit on GitHub](https://github.com/Oxygem/pytask/blob/develop/docs/index.rst)

---

pytask[¶](#pytask "Permalink to this headline")

===============================================

A simple asynchronous Python daemon for IO bound tasks, based on greenlets. Uses Redis or

Redis Cluster to store state; workers are stateless. Included `Monitor` and `Cleanup`

tasks to handle failure of workers. pytask is designed distributed and [*fault tolerant*](index.html#document-fault_tolerance).

Starting Workers[¶](#starting-workers "Permalink to this headline")

-------------------------------------------------------------------

To create a pytask worker you create an instance of `PyTask`, add your task classes to

it and then call its `run` method:

```

# Gevent monkey patching for async tasks

from gevent import monkey

monkey.patch\_all()

from pytask import PyTask

from mytasks import MyTask

# Create pytask "app"

task\_app = PyTask({'host': 'localhost', 'port': 6379})

# Add our task to it

task\_app.add\_task(MyTask)

# Run the worker

task\_app.run()

```

Writing Tasks[¶](#writing-tasks "Permalink to this headline")

-------------------------------------------------------------

The definitive guide to writing pytask tasks. The first step is to name your task, like so:

```

class MyTask(Task):

    NAME = 'mytask'

```

It is recommended to namespace your names, for example the `Monitor` and `Cleanup`

tasks included with pytask are named `pytask/monitor` and `pytask/cleanup`

respectively.

### Data & Init[¶](#data-init "Permalink to this headline")

Tasks should define an `\_\_init\_\_` function which will be passed task data as kwargs.

Remember task data is defined in JSON, so you’ll need to handle non-JSON-serializable

objects.

```

class MyTask(Task):

    def \_\_init\_\_(self, my\_task\_option='default'):

        self.option = my\_task\_option

```

### Start & Stop[¶](#start-stop "Permalink to this headline")

Tasks *must* implement a `.start` method, this is called directly in a greenlet to start

and execute the task. When this function ends the task is considered to be ended.

```

class MyTask(Task):

    ...

    def start(self):

        do\_some\_work()

```

This works well for short lived tasks, but sometimes we want to run something which runs

forever/a-long-time. To achieve this we can use a separate greenlet and the `.stop`

method to gracefully handle shutdown:

```

class MyTask(Task):

    ...

    def start(self):

        self.loop = gevent.spawn(self.work)

        self.loop.get()

    def stop(self):

        self.loop.kill()

    def work(self):

        do\_some\_work()

```

### Emitting Events[¶](#emitting-events "Permalink to this headline")

Tasks can emit events to Redis pubsub:

```

class MyTask(Task):

    ...

    def start(self):

        self.emit('event-name', {'some': 'data'})

```

### Handling Errors[¶](#handling-errors "Permalink to this headline")

When tasks encounter an error, they should raise a `self.Error` exception. This will

put them into `ERROR` state - crucially different to `EXCEPTION` which is reserved

for unexpected errors.

```

class MyTask(Task):

    ...

    def start(self):

        raise self.Error('Task failz!')

```

### Context[¶](#context "Permalink to this headline")

Sometimes it’s desirable to wrap task method calls with some kind of application specific

context (eg a Flask app). To do this a task needs to define a static `provide\_context`

method, like so:

```

class MyTask(Task):

    @staticmethod

    def provide\_context():

        # Wrap task methods in Flask.app\_context

        return web\_app.app\_context()

```

The context will be created before the task instance is created, and all task methods

(`\_\_init\_\_`, `start` & `stop`) will be nested inside the context.

pytask API[¶](#module-pytask.pytask "Permalink to this headline")

-----------------------------------------------------------------

*class* `pytask.pytask.``PyTask`(*redis\_instance*, *update\_task\_interval=5*, *\*\*kwargs*)[¶](#pytask.pytask.PyTask "Permalink to this definition")

Bases: `pytask.helpers.\_PyTaskRedisConf`

A daemon that starts/stops tasks & replicates that to a Redis instance

tasks can be control via Redis pubsub.

Redis Instance:

The first argument can be either a Redis client or a list of host/port details.

When using a list, pytask will use `redis-py` with one host and

`redis-py-cluster` where multiple hosts are present.

| Parameters: | * **redis\_instance** (*client or list*) – Redis client or list of `(host, port)` tuples

* **task\_set** (*str*) – name of task set

* **task\_prefix** (*str*) – prefix for task names

* **new\_queue** (*str*) – queue to read new task IDs from

* **end\_queue** (*str*) – where to push complete task IDs

* **update\_task\_interval** (*int*) – interval in s to update task times

 |

*exception* `PyTaskException`[¶](#pytask.pytask.PyTask.PyTaskException "Permalink to this definition")

Bases: `exceptions.Exception`

*exception* `PyTask.``StopTask`[¶](#pytask.pytask.PyTask.StopTask "Permalink to this definition")

Bases: `pytask.pytask.PyTaskException`

`PyTask.``add_exception_handler`(*handler*)[¶](#pytask.pytask.PyTask.add_exception_handler "Permalink to this definition")

Add an exception handler.

`PyTask.``add_task`(*task\_class*)[¶](#pytask.pytask.PyTask.add_task "Permalink to this definition")

Add a task class.

`PyTask.``add_tasks`(*\*task\_classes*)[¶](#pytask.pytask.PyTask.add_tasks "Permalink to this definition")

Add multiple task classes.

`PyTask.``run`(*task\_map=None*)[¶](#pytask.pytask.PyTask.run "Permalink to this definition")

Run pytask, basically a wrapper to handle KeyboardInterrupt.

`PyTask.``start_local_task`(*task\_name*, *\*\*task\_data*)[¶](#pytask.pytask.PyTask.start_local_task "Permalink to this definition")

Used to start local tasks on this worker, which will start when `.run` is

called.

### Helpers[¶](#module-pytask.helpers "Permalink to this headline")

*class* `pytask.helpers.``PyTaskHelpers`(*redis\_instance*, *task\_set='tasks'*, *task\_prefix='task-'*, *new\_queue='new-task'*, *end\_queue='end-task'*)[¶](#pytask.helpers.PyTaskHelpers "Permalink to this definition")

Bases: `pytask.helpers.\_PyTaskRedisConf`

Helper functions for managing task data within Redis. All `PyTask` instances have

an instance attached on their `helpers` attribute.

| Parameters: | * **redis\_instance** (*client or list*) – Redis client or list of `(host, port)` tuples

* **task\_set** (*str*) – name of task set

* **task\_prefix** (*str*) – prefix for task names

* **new\_queue** (*str*) – queue to read new task IDs from

* **end\_queue** (*str*) – where to push complete task IDs

 |

`get_active_task_ids`()[¶](#pytask.helpers.PyTaskHelpers.get_active_task_ids "Permalink to this definition")

Get a list of active `task\_ids`.

`get_end_task_ids`()[¶](#pytask.helpers.PyTaskHelpers.get_end_task_ids "Permalink to this definition")

Get task IDs in the end queue.

`get_new_task_ids`()[¶](#pytask.helpers.PyTaskHelpers.get_new_task_ids "Permalink to this definition")

Get task IDs in the new queue.

`get_task`(*task\_id*, *keys=None*)[¶](#pytask.helpers.PyTaskHelpers.get_task "Permalink to this definition")

Get task hash data.

`reload_task`(*task\_id*)[¶](#pytask.helpers.PyTaskHelpers.reload_task "Permalink to this definition")

Reload a task.

`remove_task`(*task\_id*)[¶](#pytask.helpers.PyTaskHelpers.remove_task "Permalink to this definition")

`restart_if_state`(*task\_id*, *states*)[¶](#pytask.helpers.PyTaskHelpers.restart_if_state "Permalink to this definition")

`restart_task`(*task\_id*)[¶](#pytask.helpers.PyTaskHelpers.restart_task "Permalink to this definition")

`set_task`(*task\_id*, *data*, *value=None*)[¶](#pytask.helpers.PyTaskHelpers.set_task "Permalink to this definition")

Set task hash data.

`start_task`(*task\_name*, *task\_id=None*, *cleanup=True*, *\*\*task\_data*)[¶](#pytask.helpers.PyTaskHelpers.start_task "Permalink to this definition")

Start a new task.

`stop_task`(*task\_id*)[¶](#pytask.helpers.PyTaskHelpers.stop_task "Permalink to this definition")

Stop a task.

`pytask.helpers.``run_loop`(*function*, *interval*)[¶](#pytask.helpers.run_loop "Permalink to this definition")

Like JavaScripts `setInterval`, slight time drift as usual, useful in long running

tasks.

### Included Tasks[¶](#module-pytask.tasks "Permalink to this headline")

*class* `pytask.tasks.``Cleanup`(*task\_handler=None*)[¶](#pytask.tasks.Cleanup "Permalink to this definition")

Bases: `pytask.task.Task`

A pytask which cleans up other pytasks! Drains the end queue and optionally triggers

a handler/callback function.

| Parameters: | **task\_handler** (*str*) – `module:attribute` type string |

`NAME` *= 'pytask/cleanup'*[¶](#pytask.tasks.Cleanup.NAME "Permalink to this definition")

`start`()[¶](#pytask.tasks.Cleanup.start "Permalink to this definition")

*class* `pytask.tasks.``Monitor`(*loop\_interval=10*, *task\_timeout=60*)[¶](#pytask.tasks.Monitor "Permalink to this definition")

Bases: `pytask.task.Task`

A pytask which monitors pytasks! Checks all task state in Redis at a configured

interval, will re-queue tasks which timeout.

`NAME` *= 'pytask/monitor'*[¶](#pytask.tasks.Monitor.NAME "Permalink to this definition")

`check_tasks`()[¶](#pytask.tasks.Monitor.check_tasks "Permalink to this definition")

`start`()[¶](#pytask.tasks.Monitor.start "Permalink to this definition")

`stop`()[¶](#pytask.tasks.Monitor.stop "Permalink to this definition")

Clustering pytask[¶](#clustering-pytask "Permalink to this headline")

---------------------------------------------------------------------

pytask leans on Redis to handle its state - this means pytask workers are stateless, and

can be scaled horizontally. Combined with a Redis cluster you have a fullly distributed and

[*somewhat fault tolerant*](index.html#document-fault_tolerance) task system. For example:

* 3x servers

* Each has one instance of Redis clustered

* Each runs 4 workers

Fault Tolerance[¶](#fault-tolerance "Permalink to this headline")

-----------------------------------------------------------------

This document describes the data model and related fault tolerance properties of pytask

instances and clusters.

### Data Model[¶](#data-model "Permalink to this headline")

* New task IDs and ended task IDs are pushed to two Redis list queues (LPUSH/RPOP)

* “Active” tasks are stored in a set

* A task is represented by a Redis hash (it’s key is the task ID w/prefix):

```

{

    'task': 'task\_name', # the NAME attribute of the task class

    'data': 'json\_data', # JSON data passed to the task

    'last\_update': 0, # Internal update counter, used to requeue failed worker tasks

    'state': '[RUNNING|STOPPED|SUCCESS|ERROR|EXCEPTION]', # tasks current state

    # Present on ended tasks:

    'output': 'json\_data' # output JSON data or exception data

}

```

At any one time, all task\_ids the system is aware of are stored in the task set, new and

end queues. Task hashes without their keys in one of these lists would be orphaned.

### Task States[¶](#task-states "Permalink to this headline")

Before task starts:

When the task is waiting to be picked by a worker - it’s ID should be in the new task

queue. Represented by no value or `WAIT`.

`RUNNING`:

These tasks are monitored and “locked” to their worker by update time. Monitor tasks

will requeue these tasks when their update time is over a configurable threshold.

`STOPPED`:

These tasks were intentionally stopped before they could complete. They can be

restarted (possibly any worker) by pushing the task ID onto the new queue.

`SUCCESS`, `ERROR` & `EXCEPTION`:

These tasks have finished, and should have an associated `output` in their hash set.

### Worker Failure[¶](#worker-failure "Permalink to this headline")

Expected/SIGINT:

The worker will stop and requeue any running tasks, and set their state to `WAIT`.

Unexpected:

The worker will stop updating `update\_time` on its tasks. Any monitor tasks will

pick this up (depending on configuration) and requeue the failed workers tasks.

### Redis Failure[¶](#redis-failure "Permalink to this headline")

Partitons:

When workers can’t reach Redis, they stop all their tasks. So long as there is still

a Redis cluster and some worker instances reaching it, these tasks will be requeued.

Complete failure:

In the case where Redis is offline for all workers, the task cluster will essentialy

stop. A usable Redis cluster is *always* required.

