tasks.py 文件源码

def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
        self._flask_app: t.Any = None
        super(MyCelery, self).__init__(*args, **kwargs)

        if t.TYPE_CHECKING:  # pragma: no cover

            class TaskBase:
                pass
        else:
            TaskBase = self.Task

        outer_self = self

        class ContextTask(TaskBase):
            abstract = True

            def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any:
                # This is not written by us but taken from here:
                # https://web.archive.org/web/20150617151604/http://slides.skien.cc/flask-hacks-and-best-practices/#15

                if outer_self._flask_app is None:  # pragma: no cover
                    raise ValueError('You forgot the initialize celery!')

                if outer_self._flask_app.config['TESTING']:
                    return TaskBase.__call__(self, *args, **kwargs)

                with outer_self._flask_app.app_context():  # pragma: no cover
                    return TaskBase.__call__(self, *args, **kwargs)

        self.Task = ContextTask