celery.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
# @Version     : Python 3.11.4
# @Software    : Sublime Text 4
# @Author      : StudentCWZ
# @Email       : StudentCWZ@outlook.com
# @Date        : 2023/11/7 17:15
# @File        : celery.py
# @Description :
"""

from celery import Celery, Task
from flask import Flask


def make_celery(app: Flask):
    celery = Celery(
        app.import_name,
        backend='redis://localhost:6379/13',
        broker='redis://localhost:6379/13'
    )
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.conf.update({
        'task_cls': ContextTask,
    })

    return celery