fantasy/cli.py
def queue(celery_arguments):
    """启动队列服务[开发中]"""

    if not app.celery:
        return click.echo(
            click.style('No celery config found，skip start...', fg='yellow'))

    celery = app.celery
    celery.autodiscover_tasks()

    argv = celery_arguments.split()
    argv.insert(0, 'worker')
    argv.insert(0, 'Queue')
    celery.worker_main(argv)
    pass
def queue(celery_arguments):
    """启动队列服务[开发中]"""

    if not app.celery:
        return click.echo(
            click.style('No celery config found，skip start...', fg='yellow'))

    celery = app.celery
    celery.autodiscover_tasks()

    argv = celery_arguments.split()
    argv.insert(0, 'worker')
    argv.insert(0, 'Queue')
    celery.worker_main(argv)
    pass
def queue ( celery_arguments ) : if not app . celery : return click . echo ( click . style ( 'No celery config found，skip start...',  f =' y ellow')) celery = app . celery celery . autodiscover_tasks ( ) argv = celery_arguments . split ( ) argv . insert ( 0 , 'worker' ) argv . insert ( 0 , 'Queue' ) celery . worker_main ( argv ) pass
启动队列服务[开发中]
启动队列服务 [ 开发中 ]