tasker.py

import collections
import types
import inspect
import functools

class Tasker(object):
    '''
    class to simplify running tasks in order
    '''
    def __init__(self):
        self.task_list = collections.deque()
    def add_single_task(self, new_task):
        '''
        add a task to the task list

        new_task must be either a function or a partial
        and that function or partial must take zero args
        '''

        #check that runs if it is a partial
        if isinstance(new_task, functools.partial):
            len_original_func_arguments    = len(inspect.getargspec(new_task.func).args)
            number_of_partialed_arguments  = len(new_task.args)

            if (len_original_func_arguments - number_of_partialed_arguments) == 0:
                self.task_list.appendleft(new_task)
            else:
                raise TypeError("Make sure that the passed partial takes no arguments")

        #check that runs if it is a pure function
        elif isinstance(new_task, types.FunctionType):
            if not inspect.getargspec(new_task).args:
                self.task_list.appendleft(new_task)
            else:
                raise TypeError("Make sure that the passed function takes no arguments")

        else:
            raise TypeError("Make sure that the passed task is a functon or partial")


    def add_multiple_tasks(self, *new_tasks):
        '''
        adds multiple tasks in order
        '''
        for task in new_tasks:
            self.add_single_task(task)
    def run_next(self):
        '''
        runs the task at the top of the deque

        if the deque is empty raises IndexError

        will not catch any exceptions the task raises

        will also return any return value of the function
        '''
        try:
            next_task = self.task_list.pop()
        except IndexError:
            raise IndexError("No more Tasks to run")

        return next_task()
    def run_all(self):
        while True:
            try:
                self.run_next()
            except IndexError:
                #expected when task list is spent
                break

test_tasker.py

'''
unittest for tasker.py
'''
import unittest
import tasker
import functools

class TestTaskerSystem(unittest.TestCase):
    '''
    runs unittests on the Tasker class
    '''
    def test_adding_single_function_with_no_args(self):
        '''
        makes sure that you can add a single function that correctly
        has no arguments
        '''
        task_manager = tasker.Tasker()
        def test_function():
            return 1

        task_manager.add_single_task(test_function)
        self.assertEqual(task_manager.run_next(), 1)
    def test_adding_multiple_functions_with_no_args(self):
        '''
        adds multiple functions that have no arguments and makes sure
        that the tasks are run in the correct order
        '''
        task_manager = tasker.Tasker()
        def test_function_1():
            return 1
        def test_function_2():
            return 2

        task_manager.add_multiple_tasks(test_function_1,
                                        test_function_2,
                                        test_function_1,
                                        test_function_1)

        expected_values = [1, 2, 1, 1]
        actual_values   = []

        actual_values.append(task_manager.run_next())
        actual_values.append(task_manager.run_next())
        actual_values.append(task_manager.run_next())
        actual_values.append(task_manager.run_next())

        self.assertEqual(expected_values,
                         actual_values)

    def test_adding_single_function_with_args(self):
        '''
        adds a function that has arguments and makes sure a TypeError is
        raised
        '''

        task_manager = tasker.Tasker()

        def function_with_args(argument):
            return argument

        with self.assertRaises(TypeError):
            task_manager.add_single_task(function_with_args)

    def test_adding_multiple_functions_with_args(self):
        '''
        adds multiple functions, one of which takes arguments,
        and makes sure that a TypeError is raised
        '''
        task_manager = tasker.Tasker()

        def function_without_args():
            return 1

        def function_with_args(argument):
            return argument

        with self.assertRaises(TypeError):
            task_manager.add_multiple_tasks(function_without_args,
                                            function_with_args,
                                            function_without_args)

    def test_adding_single_partial_without_args(self):
        '''
        adds a single partial that correctly has no arguments
        '''

        task_manager = tasker.Tasker()

        def function_with_args(argument):
            return argument

        #this partial should return 1 with no args
        partial_function = functools.partial(function_with_args, 1)


        task_manager.add_single_task(partial_function)
        self.assertEqual(task_manager.run_next(), 1)

    def test_adding_multiple_partials_with_no_args(self):
        '''
        adds multiple partials that have no arguments and makes sure
        that the tasks are run in the correct order
        '''
        task_manager = tasker.Tasker()

        def function_with_args(argument):
            return argument

        test_partial_1 = functools.partial(function_with_args, 1)
        test_partial_2 = functools.partial(function_with_args, 2)

        task_manager.add_multiple_tasks(test_partial_1,
                                        test_partial_2,
                                        test_partial_1,
                                        test_partial_1)

        expected_values = [1, 2, 1, 1]
        actual_values   = []

        actual_values.append(task_manager.run_next())
        actual_values.append(task_manager.run_next())
        actual_values.append(task_manager.run_next())
        actual_values.append(task_manager.run_next())

        self.assertEqual(expected_values,
                         actual_values)

    def test_adding_single_partial_with_args(self):
        '''
        adds a single partial that incorrectly has unfilled arguments
        and makes sure a TypeError is raised
        '''

        task_manager = tasker.Tasker()

        def function_with_two_args(argument_1, argument_2):
            return argument_1 + argument_2

        #this partial should return 1 with no args
        partial_function = functools.partial(function_with_two_args, 1)


        with self.assertRaises(TypeError):
            task_manager.add_single_task(partial_function)