[GitHub] [airflow] turbaszek commented on a change in pull request #7542: [AIRFLOW-6921] Fetch celery states in bulk

[GitBox <gi...@apache.org>]

turbaszek commented on a change in pull request #7542:
URL: https://github.com/apache/airflow/pull/7542#discussion_r420705597



##########
File path: airflow/executors/celery_executor.py
##########
@@ -319,3 +262,109 @@ def execute_async(self,
 
     def terminate(self):
         pass
+
+
+def fetch_celery_task_state(async_result: AsyncResult) -> Tuple[str, Union[str, ExceptionWithTraceback]]:
+    """
+    Fetch and return the state of the given celery task. The scope of this function is
+    global so that it can be called by subprocesses in the pool.
+
+    :param async_result: a tuple of the Celery task key and the async Celery object used
+        to fetch the task's state
+    :type async_result: tuple(str, celery.result.AsyncResult)
+    :return: a tuple of the Celery task key and the Celery state of the task
+    :rtype: tuple[str, str]
+    """
+
+    try:
+        with timeout(seconds=OPERATION_TIMEOUT):
+            # Accessing state property of celery task will make actual network request
+            # to get the current state of the task
+            return async_result.task_id, async_result.state
+    except Exception as e:  # pylint: disable=broad-except
+        exception_traceback = "Celery Task ID: {}\n{}".format(async_result, traceback.format_exc())

Review comment:
       ```suggestion
           exception_traceback = f"Celery Task ID: {async_result}\n{traceback.format_exc()}"
   ```




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org