docstr: Pulls the status of each Work instance in the cloud.
  
  Normally, the Lightning frameworks communicates statuses through the queues, but while the Work instance is
  being provisionied, the queues don't exist yet and hence we need to make API calls directly to the backend to
  fetch the status and update it in the states.
docstr_tok: ["Pulls","the","status","of","each","Work","instance","in","the","cloud",".","Normally",",","the","Lightning","frameworks","communicates","statuses","through","the","queues",",","but","while","the","Work","instance","is","being","provisionied",",","the","queues","do","n't","exist","yet","and","hence","we","need","to","make","API","calls","directly","to","the","backend","to","fetch","the","status","and","update","it","in","the","states","."]
code: def update_work_statuses(self, works: List[LightningWork]) -> None:
      """Pulls the status of each Work instance in the cloud.
  
      Normally, the Lightning frameworks communicates statuses through the queues, but while the Work instance is
      being provisionied, the queues don't exist yet and hence we need to make API calls directly to the backend to
      fetch the status and update it in the states.
  
      """
      if not works:
          return
  
      # TODO: should this run in a timer thread instead?
      if (
          self._last_time_updated is not None
          and monotonic() - self._last_time_updated
          < self._status_update_interval
      ):
          return
  
      cloud_work_specs = self._get_cloud_work_specs(self.client)
      local_works = works
      for cloud_work_spec in cloud_work_specs:
          for local_work in local_works:
              # TODO (tchaton) Better resolve pending status after succeeded
  
              # 1. Skip if the work isn't the current one.
              if local_work.name != cloud_work_spec.name:
                  continue
  
              # 2. Logic for idle timeout
              self._handle_idle_timeout(
                  local_work.cloud_compute.idle_timeout,
                  local_work,
                  cloud_work_spec,
              )
  
              # 3. Map the cloud phase to the local one
              cloud_stage = cloud_work_stage_to_work_status_stage(
                  cloud_work_spec.status.phase,
              )
  
              # 4. Detect if the work failed during pending phase
              if (
                  local_work.status.stage == WorkStageStatus.PENDING
                  and cloud_stage in WorkStageStatus.FAILED
              ):
                  if local_work._raise_exception:
                      raise Exception(
                          f"The work {local_work.name} failed during pending phase."
                      )
                  logger.error(
                      f"The work {local_work.name} failed during pending phase."
                  )
  
              # 5. Skip the pending and running as this is already handled by Lightning.
              if cloud_stage in (
                  WorkStageStatus.PENDING,
                  WorkStageStatus.RUNNING,
              ):
                  continue
  
              # TODO: Add the logic for wait_timeout
              if local_work.status.stage != cloud_stage:
                  latest_hash = local_work._calls["latest_call_hash"]
                  if latest_hash is None:
                      continue
                  local_work._calls[latest_hash]["statuses"].append(
                      make_status(cloud_stage)
                  )
  
      self._last_time_updated = monotonic()
  
imports: import inspect
  import json
  import logging
  import os
  import random
  import string
  import urllib
  from time import monotonic, sleep, time
  from typing import List, Optional
  from lightning_cloud.openapi import AppinstancesIdBody, Externalv1LightningappInstance, Externalv1Lightningwork, V1BuildSpec, V1Drive, V1DriveSpec, V1DriveStatus, V1DriveType, V1Flowserver, V1LightningappInstanceState, V1LightningappRestartPolicy, V1LightningworkClusterDriver, V1LightningworkDrives, V1LightningworkSpec, V1LightningworkState, V1ListLightningworkResponse, V1Metadata, V1NetworkConfig, V1PackageManager, V1PythonDependencyInfo, V1SourceType, V1UserRequestedComputeConfig
  from lightning_cloud.openapi.rest import ApiException
  from lightning.app.core import LightningApp, LightningWork
  from lightning.app.core.queues import QueuingSystem
  from lightning.app.runners.backends.backend import Backend
  from lightning.app.storage import Drive, Mount
  from lightning.app.utilities.enum import WorkStageStatus, WorkStopReasons, make_status
  from lightning.app.utilities.exceptions import LightningPlatformException
  from lightning.app.utilities.network import LightningClient, _check_service_url_is_ready
  from lightning_cloud.openapi import SpecLightningappInstanceIdWorksBody, WorksIdBody
label_desc: Use image node_id 4 for calling the CloudBackend obj's underlying member method code with example usage: obj.update_work_statuses(works) without return types