safe_celery

circle/common/safe_celery_get.py

+# safe_celery_get.py
+"""
+Drop-in replacement for Celery's .get() that works with mixed Celery 2.7 / 3 / 4 workers.
+Handles all result formats: raw Exception, Celery meta-dict, JSON, pickle, pickle_v2.
+"""
+
+class RemoteTaskError(Exception):
+    """Unified remote task failure exception (Celery 3/4 compatible)."""
+
+    def __init__(self, exc_type, message, traceback=None):
+        self.exc_type = exc_type
+        self.remote_message = message
+        self.remote_traceback = traceback
+
+        # Python 2.7: no f-strings
+        text = "[{0}] {1}".format(exc_type, message)
+
+        # Python 2.7-compatible super()
+        super(RemoteTaskError, self).__init__(text)
+
+
+def safe_celery_get(async_result, timeout=None):
+    """
+    Celery 3 and 4 compatible .get() replacement.
+
+    - Always calls get(propagate=False) so Celery never raises dict.
+    - Interprets Celery meta info manually.
+    - Reconstructs clean Python Exception for caller.
+
+    :param async_result: Celery AsyncResult object
+    :param timeout: timeout for result.wait/get
+    :return: task result or raises RemoteTaskError / Exception
+    """
+
+    # Prevent Celery from raising exceptions on our behalf
+    _ = async_result.get(timeout=timeout, propagate=False)
+
+    # Load full metadata
+    meta = async_result.backend.get_task_meta(async_result.id)
+    status = meta.get("status")
+    result = meta.get("result")
+
+    # SUCCESS return payload
+    if status == "SUCCESS":
+        return result
+
+    # FAILURE handling
+    # Case 1: If worker sent actual Exception (possible on Celery 3 + pickle)
+    if isinstance(result, BaseException):
+        raise result
+
+    # Case 2: Celery meta-dict with exc_type + exc_message (Celery 3 and 4)
+    if isinstance(result, dict) and "exc_type" in result and "exc_message" in result:
+        exc_type = result.get("exc_type") or "RemoteError"
+        exc_message = result.get("exc_message")
+
+        # e.g. ('msg',) or ['msg']
+        if isinstance(exc_message, (list, tuple)):
+            if exc_message:
+                exc_message = exc_message[0]
+            else:
+                exc_message = "<empty remote message>"
+
+        traceback = meta.get("traceback")
+        raise RemoteTaskError(exc_type, exc_message, traceback)
+
+    # Case 3: Some unknown format
+    raise Exception("Remote task failed with unknown result format: {0!r}".format(result))
+