task tree

tasktree.py

import sys
from celery import Celery
from celery.canvas import group
from kombu import Queue, Exchange

celery = Celery(
    broker="amqp://test:test@localhost:5672/test"
)
celery.conf.update(
    CELERY_RESULT_BACKEND = "mongodb",
)

@celery.task
def task_a(result):
    print 'task_a:', result
    return result

@celery.task
def task_b(result):
    print 'task_b:', result
    return result

@celery.task
def task_c(result):
    print 'task_c:', result
    return result

@celery.task
def notify_user(result):
    print result
    return result


if __name__ == '__main__':
    if len(sys.argv) > 1 and sys.argv[1] == 'produce-test':

        tree = [
            [["C1", "C2", "C3"], ["C4", "C5"]], [["C6", "C7", "C8"], ["C9"]]
        ]
        a_group = []
        for ia, a in enumerate(tree):
            print "A%s:" % ia
            b_group = []
            for ib, b in enumerate(a):
                print " - B%s:" % ib
                for c in b:
                    print '   -', c

                c_group = group([task_c.s(c) for c in b])

                b_group.append(c_group | task_b.s())

            a_group.append(group(b_group) | task_a.s())

        final_task = group(a_group) | notify_user.s()
        print final_task
        final_result = final_task()
        print final_result.get()
    else:
        celery.start()