functastic 2.1

tasks which never succeed (on purpose)

from functastic import Task
# if you want a task to run over and over forever, say 10 seconds after
# it last finished, set the success condition to `lambda t: None`
# and do not specify attempts
task = Task(f, args=['a'], delay=10,
            success_condition=lambda t: None)

# this task will be run 5 times, 10 seconds after the previous run,
# regardless of what f returns or raises
task = Task(f, args=['a'], delay=10, attempts=5,
            success_condition=lambda t: None)

tasks which “succeed” by failing in some way

sometimes it makes sense to think of success_condition as a break condition. let’s say there is a function that makes requests to a webservice. it makes sense to retry if you get a 404 because you are polling for a resource to become available, but if you get another error, like 401 unauthorized, retrying wouldn’t make sense, so we want to break (or “succeed”) if we get a non 404 exception:

import requests
from functastic import Task

def handle_resource(res):
    pass

def get_thing(url):
    r = requests.get(url)
    r.raise_for_status()
    handle_resource(r.json())

# every 5 seconds attempt to fetch the resource and handle it,
# quit when there is no exception (actual success) or there is
# a non 404 exception (call failed in a bad way, so don't try it again)
task = Task(get_thing, args['http://whatever.com/resource/id'], delay=5,
            success_condition=lambda t: (t.exception and '404' not in t.exception.message or
                                         not t.exception))

using tasks without TaskHeap

from functastic import Task

task = Task(f, args['a'], attempts=10)

# task.retry always starts as True
while task.retry:
    # calling the task calls the function
    task()
    # at this point, task.retry may have become False depending on the task
    # optionally sleep in between calls
    time.sleep(2)

unthreaded use

run a task or set of tasks and return when they finish. without stop=True the tasks.run() call will block forever because it won’t stop iterating every TaskHeap interval.

from functastic import Task
from functastic import TaskHeap
# add tasks and then run run(stop=True)
tasks = TaskHeap()
tasks.append(Task(usually_fails, args=['a'], delay=1))
tasks.append(Task(usually_fails, args=['b'], attempts=10, delay=1))
tasks.run(stop=True)

use with threading library

TaskHeap works well with threading libraries. this will run the task loop in another thread and add tasks willy nilly while they run

import eventlet
from functastic import Task
from functastic import TaskHeap
# note the use of eventlet.sleep here to specify which sleep
# function TaskHeap should use, or use monkey patching
# interval can also be passed if you don't like the default 0.1s
# this sets the interval task run interval to 3 seconds
tasks = TaskHeap(sleep=eventlet.sleep, interval=3)
eventlet.spawn(tasks.run)
tasks.append(Task(usually_fails, args=['a'], delay=1))
tasks.append(Task(usually_fails, args=['b'], attempts=10, delay=1))

# have to sleep here to surrender execution to the green thread
while tasks:
    tasks.sleep()

stopping tasks

once a TaskHeap has been started with run(), it will run indefinitely unless stop=True is passed in run(stop=True). it can be stopped in two different ways: - stop_after(), causes the task loop to exit once all tasks are completed - stop_now(), causes the task loop to stop as soon as possible. since the task loop is single threaded, it will only exit after finishing the current iteration. this means the current task, if there is one, will continue as planned, but all future tasks will not be run unless run() is called again.

import eventlet
from functastic import Task
from functastic import TaskHeap

tasks = TaskHeap(sleep=eventlet.sleep)
gt = eventlet.spawn(tasks.run)
tasks.append(Task(usually_fails, args=['a'], delay=1))
tasks.append(Task(usually_fails, args=['b'], attempts=10, delay=1))

# stop the tasks thread after 5 second, gt.wait() will return almost
# instantly
tasks.sleep(5)
tasks.stop_now()
gt.wait()             # <-- this line should return quickly

# start the tasks again
gt = eventlet.spawn(tasks.run)

# this time tell the tasks loop to exit once finished
tasks.sleep()
tasks.stop_after()
gt.wait()             # <-- this line should return when all tasks complete