class_docstr: Abstract subclass of NodeProvider meant for use with external cluster managers.
  
  Batches reads of cluster state into a single method, get_node_data, called at the
  start of an autoscaling update.
  
  Batches modifications to cluster state into a single method, submit_scale_request,
  called at the end of an autoscaling update.
  
  Implementing a concrete subclass of BatchingNodeProvider only requires overriding
  get_node_data() and submit_scale_request().
  
  See the method docstrings for more information.
  
  Note that an autoscaling update may be conditionally
  cancelled using the optional method safe_to_scale()
  of the root NodeProvider.
code: def __init__(
      self,
      provider_config: Dict[str, Any],
      cluster_name: str,
      _allow_multiple: bool = False,
  ) -> None:
      NodeProvider.__init__(self, provider_config, cluster_name)
      self.node_data_dict: Dict[NodeID, NodeData] = {}
  
      # Disallow multiple node providers, unless explicitly allowed for testing.
      global provider_exists
      if not _allow_multiple:
          assert (
              not provider_exists
          ), "Only one BatchingNodeProvider allowed per process."
  
      # These flags enforce correct behavior for single-threaded node providers
      # which interact with external cluster managers:
      assert (
          provider_config.get(DISABLE_NODE_UPDATERS_KEY, False) is True
      ), f"To use BatchingNodeProvider, must set `{DISABLE_NODE_UPDATERS_KEY}:True`."
      assert (
          provider_config.get(DISABLE_LAUNCH_CONFIG_CHECK_KEY, False)
          is True
      ), (
          "To use BatchingNodeProvider, must set "
          f"`{DISABLE_LAUNCH_CONFIG_CHECK_KEY}:True`."
      )
      assert (
          provider_config.get(FOREGROUND_NODE_LAUNCH_KEY, False) is True
      ), f"To use BatchingNodeProvider, must set `{FOREGROUND_NODE_LAUNCH_KEY}:True`."
  
      # self.scale_change_needed tracks whether we need to update scale.
      # set to True in create_node and terminate_nodes calls
      # reset to False in non_terminated_nodes, which occurs at the start of the
      #   autoscaling update. For good measure, also set to false in post_process.
      self.scale_change_needed = False
  
      self.scale_request = ScaleRequest()
  
imports: import logging
  from collections import defaultdict
  from dataclasses import dataclass, field
  from typing import Any, Dict, List, Optional, Set
  from ray.autoscaler._private.constants import DISABLE_LAUNCH_CONFIG_CHECK_KEY, DISABLE_NODE_UPDATERS_KEY, FOREGROUND_NODE_LAUNCH_KEY
  from ray.autoscaler._private.util import NodeID, NodeIP, NodeKind, NodeStatus, NodeType
  from ray.autoscaler.node_provider import NodeProvider
  from ray.autoscaler.tags import NODE_KIND_HEAD, TAG_RAY_NODE_KIND, TAG_RAY_NODE_STATUS, TAG_RAY_USER_NODE_TYPE
label_desc: Use image node_id 1 to create a new BatchingNodeProvider object from inherited base classes: NodeProvider with example: obj = BatchingNodeProvider(provider_config, cluster_name, _allow_multiple)