celerystalk.io - Celery Class Tasks

base_task.py

from celery import Celery

app = Celery('NAME', broker='BROKER')

class BaseTask(app.Task):
		def on_failure(self, exc, task_id, args, kwargs, einfo):
				logging_tool.log(exc, task_id, args, kwargs, einfo)

base_task_decorator.py

from base_task import BaseTask

@app.task(base=BaseTask)
def add(x, y):
		return x + y

base_task_decorator_2.py

from celery import Celery

app = Celery('NAME', task_cls='your.module.path:BaseTask')

decorator.py

from celery import Celery

app = Celery('NAME', broker='BROKER')

@task(rate_limit='100/s')
def task():
    pass

inherit_1.py

from celery import Celery
import time

app = Celery('NAME', broker='BROKER')

class Count():
    def __init__(self):
        self.pos = 1

    def counter(self):
        self.pos += 1


class Counter(app.Task, Count):
    def run(self, wait_time):
        time.sleep(wait_time)
        self.counter()
				print(self.pos)

    
CounterTask = app.register_task(Counter())

inherit_2.py

from celery import Celery
import time

app = Celery('NAME', broker='BROKER')

class Count():
    def __init__(self):
        self.pos = 1

    def counter(self):
        self.pos += 1


class Counter(app.Task):    
    def run(self, wait_time):
        self.count_handler = Count()
        time.sleep(wait_time)
        self.count_handler.counter()
        print(self.count_handler.pos)

    
CounterTask = app.register_task(Counter())

profile_create.py

from celery import Celery

app = Celery('NAME', broker='BROKER')

class ProfileCreate(app.Task):
    def run(self, user_id, name, email, avatar):
        self.add_email_to_mailing_list(email)
				self.upload_avatar_to_s3(avatar)
			
		def add_email_to_mailing_list(self, email):
				mailchimp.add_to_list(email)
	
		def upload_avatar_to_s3(self, avatar):
				s3.upload(avatar)

ProfileCreateTask = app.register_task(ProfileCreate())

profile_create_execution.py

ProfileCreateTask.delay((42, "Tony Stark", "[email protected]", image_data))

profile_create_with_base_task.py

from base_task import BaseTask

class ProfileCreate(BaseTask):
    
    def run(self, user_id, first_name, last_name, email, avatar):
        self.add_email_to_mailing_list(email)
				self.upload_avatar_to_s3(avatar)
			
		def add_email_to_mailing_list(self, email):
				mailchimp.add_to_list(email)
	
		def upload_avatar_to_s3(self, avatar):
				s3.upload(avatar)

ProfileCreateTask = app.register_task(ProfileCreate())