code: def fetch_celery_task_state(
      async_result: AsyncResult,
  ) -> tuple[str, str | ExceptionWithTraceback, Any]:
      """
      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
      :return: a tuple of the Celery task key and the Celery state and the celery info
          of the task
      """
      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
              info = (
                  async_result.info
                  if hasattr(async_result, "info")
                  else None
              )
              return async_result.task_id, async_result.state, info
      except Exception as e:
          exception_traceback = f"Celery Task ID: {async_result}\n{traceback.format_exc()}"
          return (
              async_result.task_id,
              ExceptionWithTraceback(e, exception_traceback),
              None,
          )
  
imports: from __future__ import annotations
  import contextlib
  import logging
  import math
  import os
  import subprocess
  import traceback
  import warnings
  from concurrent.futures import ProcessPoolExecutor
  from typing import TYPE_CHECKING, Any, Mapping, MutableMapping, Optional, Tuple
  from celery import Celery, Task, states
  from celery.backends.base import BaseKeyValueStoreBackend
  from celery.backends.database import DatabaseBackend, Task, retry, session_cleanup
  from celery.signals import import_modules
  from setproctitle import setproctitle
  from sqlalchemy import select
  import airflow.settings
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException, RemovedInAirflow3Warning
  from airflow.executors.base_executor import BaseExecutor
  from airflow.stats import Stats
  from airflow.utils.dag_parsing_context import _airflow_parsing_context_manager
  from airflow.utils.log.logging_mixin import LoggingMixin
  from airflow.utils.net import get_hostname
  from airflow.utils.providers_configuration_loader import providers_configuration_loaded
  from airflow.utils.timeout import timeout
label_desc: Use image node_id 7 for calling a global function with example usage: fetch_celery_task_state(async_result) and returns: async_result,  async_result,  info, async_result,  ExceptionWithTraceback,  None