scripts/utils.py

import time
from celery import Celery
from celery.result import AsyncResult
from typing import Dict, Any


# Wait for the celery result
def wait_for_result(app: Celery, task_id: str, timeout: float = 120.0, poll: float = 0.5) -> Dict[str, Any]:
    """
    Poll for a result with a timeout. If task updates state with meta (e.g., PROGRESS),
    we surface that along the way.
    """
    res = AsyncResult(task_id, app=app)
    t0 = time.time()
    last_state = None

    while True:
        state = res.state
        if state != last_state:
            print(f"State: {state} | Info: {res.info}")
            last_state = state

        if res.ready():
            # could be SUCCESS or FAILURE; .get() will raise on FAILURE
            return res.get(propagate=False)  # returns exception object if failed

        if time.time() - t0 > timeout:
            raise TimeoutError(f"Task {task_id} did not finish in {timeout} seconds.")
        time.sleep(poll)