File size: 959 Bytes
25f9bfc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 | 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) |