Как получить stdout реального времени из задачи celery?

У меня есть проект Django, в котором я запускаю несколько фоновых задач с помощью celery, одна из моих целей - показать stdout запущенной задачи в реальном времени, чтобы пользователь мог видеть прогресс. Проблема в том, что я получаю stdout процесса только тогда, когда он завершает выполнение (все сбрасывается сразу).

Views.py

def run_script(request, file_type, file_id):
    run_features.delay(arguments)
    return

Tasks.py

@shared_task
def run_features(arguments):
    process = subprocess.Popen(arguments, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)

        channel_layer = get_channel_layer()

        while True:
            output = process.stdout.readline()
            if output == '' and process.poll() is not None:
                break
            if output:
                print(output)
                async_to_sync(channel_layer.group_send)(
                    'output_group',
                    {
                        'type': 'send_output',
                        'output': output.strip()
                    }
                )

        process.wait()
        return process.returncode

Celery INFO

 -------------- celery@ritik v5.3.1 (emerald-rush)
--- ***** ----- 
-- ******* ---- Linux-6.5.0-25-generic-x86_64-with-glibc2.35 2024-03-21 09:30:54
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         Automation:0x7e2bb4073280
- ** ---------- .> transport:   redis://localhost:6379/0
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery

Ожидание - я получу stdout, строка за строкой, во время выполнения задачи. Фактический результат - я получаю stdout, весь сразу, после завершения подпроцесса.

Кроме того, я не уверен, является ли это проблемой django или проблемой celery.

Спасибо за любую помощь! Это мой первый вопрос на stackOverflow.

Найдено решение! Просто используйте Pexpect.spawn() вместо subprocess.Popen()

# Start the process
child = pexpect.spawn(command, args, encoding='utf-8', timeout=180)

# Log the output
while True:
    try:
        line = child.readline()
        if not line:
            break
        print(line.strip())
    except pexpect.EOF:
        break

child.wait()
child.close()

return None
Вернуться на верх