将额外的属性添加到可从自定义基类访问的celery函数

from celery import Celery, Task

app = Celery('tasks', broker='pyamqp://guest@localhost//')

class BaseTask(Task):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.custom_attribute = "This is a custom attribute"

@app.task(base=BaseTask)
def my_task():
    print(f"Custom attribute value: {my_task.custom_attribute}")
    # 这里可以添加任务的具体逻辑

# 调用任务
my_task.delay()

class BaseTask(Task):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.custom_attribute = "Custom Value"

class MyTask(BaseTask):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)  # 确保调用基类的初始化方法
        # 可以在这里添加更多自定义属性

    def run(self):
        print(f"Custom attribute value: {self.custom_attribute}")
        # 这里可以添加任务的具体逻辑