fantasy/__init__.py
def load_tasks(app, entry_file=None):
    """装载任务，解决celery无法自动装载的问题"""
    from celery import Task
    tasks_txt = os.path.join(os.path.dirname(entry_file), 'migrations',
                             'tasks.txt')

    if not os.path.exists(tasks_txt):
        import sys
        print('Tasks file not found:%s' % tasks_txt)
        sys.exit(-1)

    class ContextTask(Task):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return super().__call__(*args, **kwargs)

    app.celery.config_from_object(app.config, namespace='CELERY')
    app.celery.Task = ContextTask

    with app.app_context():
        with open(tasks_txt, 'r') as f:
            for line in f:
                mod = line.strip('\n')
                if mod:
                    importlib.import_module(mod + '.tasks')
                pass
            pass
        pass
    pass
def load_tasks(app, entry_file=None):
    """装载任务，解决celery无法自动装载的问题"""
    from celery import Task
    tasks_txt = os.path.join(os.path.dirname(entry_file), 'migrations',
                             'tasks.txt')

    if not os.path.exists(tasks_txt):
        import sys
        print('Tasks file not found:%s' % tasks_txt)
        sys.exit(-1)

    class ContextTask(Task):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return super().__call__(*args, **kwargs)

    app.celery.config_from_object(app.config, namespace='CELERY')
    app.celery.Task = ContextTask

    with app.app_context():
        with open(tasks_txt, 'r') as f:
            for line in f:
                mod = line.strip('\n')
                if mod:
                    importlib.import_module(mod + '.tasks')
                pass
            pass
        pass
    pass
装载任务，解决celery无法自动装载的问题