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
def load_tasks ( app , entry_file = None ) : 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无法自动装载的问题
装载任务，解决celery无法自动装载的问题