Celery-push

tasks.py

from celery import Celery, shared_task
import time


#Redis Result backend config:
app = Celery('tasks', broker='amqp://localhost//', backend='redis://localhost:6379/0')
#Mysql Result backend config:
# app = Celery('tasks', broker='amqp://localhost//', backend='db+mysql://root:python098@localhost/test')


@app.task(max_retries=None, shared=False)
def add(x,y):
	print "Adding..."
	time.sleep(5)
	return x+y


# BOUND TASK:
# A task being bound means the first argument to the task will always be the task instance (self), just like Python 
# bound methods.

@app.task(bind=True)
def sub(self,x,y):
	print self.request.id
	print "subracting"
	return x-y


#TASK INHERITANCE: (The BASE argument)
import celery
class MyTask(celery.Task):

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print('{0!r} failed: {1!r}'.format(task_id, exc))

@app.task(base=MyTask)
def baseAdd(x, y):
    print "This is an Inherited TASK!"	
    # raise KeyError()

#SHARED_TASK
# The @shared_task decorator lets you create tasks without having any concrete app instance in Django:

@shared_task
def reverse(name):
	return name[::-1]


@app.task(max_retries=None, shared=False)
def sum_task(alist):
	return reduce(lambda x,y: x+y, alist)	

worker.py

from tasks import *
from celery import chain, chord, group
# add.delay(21,21)



# add.apply_async((2,2),link=add.s(15),expires=60,retry=False)	

#LINK: acts like a pipe. The result of the previous call is the default parameter to the next function call. 
#Here o/p is 4, link adds 15 to 4. Result is 19.
#EXPIRES: in 60sec
#RETRY: try again if failure.


# delay(*args,**kwargs) and calling(__call__) can also be called in place of apply_async()
# However, apply_async lets you pass the most arguments.


# BOUND TASK example:
# sub.delay(22,2)


#TASK INHERITANCE ex:
# baseAdd.delay(2,2)

#SHARED_TASK ex:
# reverse.delay('Celery is awesome!')

#CELERY CHAIN:
# tasks are chained together. Each task follows the other by being applied as a callback to the previous task.

result = chain(add.s(4,4), add.s(5), add.s(10))()

#After result backend configuration
# print(result.status)   
# print(result.ready)
# print "RESULT: %s" %result.get()

#CELERY GROUP:
# creates a group of tasks that are executed in parallel
# mygroup = group([add.s(1,1), add.s(2,2)])
# promise = mygroup()

#NOTE: 
# celery -A tasks worker --loglevel=info