code: def batch(
      _func: Optional[Callable] = None,
      max_batch_size: int = 10,
      batch_wait_timeout_s: float = 0.0,
      *,
      batch_queue_cls: Type[_BatchQueue] = _BatchQueue,
  ):
      """Converts a function to asynchronously handle batches.
  
      The function can be a standalone function or a class method. In both
      cases, the function must be `async def` and take a list of objects as
      its sole argument and return a list of the same length as a result.
  
      When invoked, the caller passes a single object. These will be batched
      and executed asynchronously once there is a batch of `max_batch_size`
      or `batch_wait_timeout_s` has elapsed, whichever occurs first.
  
      `max_batch_size` and `batch_wait_timeout_s` can be updated using setter
      methods from the batch_handler (`set_max_batch_size` and
      `set_batch_wait_timeout_s`).
  
      Example:
  
      .. code-block:: python
  
              from ray import serve
              from starlette.requests import Request
  
              @serve.deployment
              class BatchedDeployment:
                  @serve.batch(max_batch_size=10, batch_wait_timeout_s=0.1)
                  async def batch_handler(self, requests: List[Request]) -> List[str]:
                      response_batch = []
                      for r in requests:
                          name = (await requests.json())["name"]
                          response_batch.append(f"Hello {name}!")
  
                      return response_batch
  
                  def update_batch_params(self, max_batch_size, batch_wait_timeout_s):
                      self.batch_handler.set_max_batch_size(max_batch_size)
                      self.batch_handler.set_batch_wait_timeout_s(batch_wait_timeout_s)
  
                  async def __call__(self, request: Request):
                      return await self.batch_handler(request)
  
              app = BatchedDeployment.bind()
  
      Arguments:
          max_batch_size: the maximum batch size that will be executed in
              one call to the underlying function.
          batch_wait_timeout_s: the maximum duration to wait for
              `max_batch_size` elements before running the current batch.
          batch_queue_cls: the class to use for the underlying batch queue.
      """
      # `_func` will be None in the case when the decorator is parametrized.
      # See the comment at the end of this function for a detailed explanation.
      if _func is not None:
          if not callable(_func):
              raise TypeError(
                  "@serve.batch can only be used to decorate functions or methods."
              )
  
          if not iscoroutinefunction(_func):
              raise TypeError(
                  "Functions decorated with @serve.batch must be 'async def'"
              )
  
      _validate_max_batch_size(max_batch_size)
      _validate_batch_wait_timeout_s(batch_wait_timeout_s)
  
      def _batch_decorator(_func):
          lazy_batch_queue_wrapper = _LazyBatchQueueWrapper(
              max_batch_size,
              batch_wait_timeout_s,
              _func,
              batch_queue_cls,
          )
  
          async def batch_handler_generator(
              first_future: asyncio.Future,
          ) -> AsyncGenerator:
              """Generator that handles generator batch functions."""
  
              future = first_future
              while True:
                  try:
                      async_response: _GeneratorResult = await future
                      future = async_response.next_future
                      yield async_response.result
                  except StopAsyncIteration:
                      break
  
          def enqueue_request(args, kwargs) -> asyncio.Future:
              self = extract_self_if_method_call(args, _func)
              flattened_args: List = flatten_args(
                  extract_signature(_func), args, kwargs
              )
  
              if self is None:
                  # For functions, inject the batch queue as an
                  # attribute of the function.
                  batch_queue_object = _func
              else:
                  # For methods, inject the batch queue as an
                  # attribute of the object.
                  batch_queue_object = self
                  # Trim the self argument from methods
                  flattened_args = flattened_args[2:]
  
              batch_queue = lazy_batch_queue_wrapper.queue
  
              # Magic batch_queue_object attributes that can be used to change the
              # batch queue attributes on the fly.
              # This is purposefully undocumented for now while we figure out
              # the best API.
              if hasattr(
                  batch_queue_object, "_ray_serve_max_batch_size"
              ):
                  new_max_batch_size = getattr(
                      batch_queue_object, "_ray_serve_max_batch_size"
                  )
                  _validate_max_batch_size(new_max_batch_size)
                  batch_queue.max_batch_size = new_max_batch_size
  
              if hasattr(
                  batch_queue_object, "_ray_serve_batch_wait_timeout_s"
              ):
                  new_batch_wait_timeout_s = getattr(
                      batch_queue_object,
                      "_ray_serve_batch_wait_timeout_s",
                  )
                  _validate_batch_wait_timeout_s(
                      new_batch_wait_timeout_s
                  )
                  batch_queue.batch_wait_timeout_s = (
                      new_batch_wait_timeout_s
                  )
  
              future = get_or_create_event_loop().create_future()
              batch_queue.put(
                  _SingleRequest(self, flattened_args, future)
              )
              return future
  
          # TODO (shrekris-anyscale): deprecate batch_queue_cls argument and
          # convert batch_wrapper into a class once `self` argument is no
          # longer needed in `enqueue_request`.
          @wraps(_func)
          def generator_batch_wrapper(*args, **kwargs):
              first_future = enqueue_request(args, kwargs)
              return batch_handler_generator(first_future)
  
          @wraps(_func)
          async def batch_wrapper(*args, **kwargs):
              # This will raise if the underlying call raised an exception.
              return await enqueue_request(args, kwargs)
  
          if isasyncgenfunction(_func):
              wrapper = generator_batch_wrapper
          else:
              wrapper = batch_wrapper
  
          # We store the lazy_batch_queue_wrapper's getters and setters as
          # batch_wrapper attributes, so they can be accessed in user code.
          wrapper._get_max_batch_size = (
              lazy_batch_queue_wrapper.get_max_batch_size
          )
          wrapper._get_batch_wait_timeout_s = (
              lazy_batch_queue_wrapper.get_batch_wait_timeout_s
          )
          wrapper.set_max_batch_size = (
              lazy_batch_queue_wrapper.set_max_batch_size
          )
          wrapper.set_batch_wait_timeout_s = (
              lazy_batch_queue_wrapper.set_batch_wait_timeout_s
          )
  
          return wrapper
  
      # Unfortunately, this is required to handle both non-parametrized
      # (@serve.batch) and parametrized (@serve.batch(**kwargs)) usage.
      # In the former case, `serve.batch` will be called with the underlying
      # function as the sole argument. In the latter case, it will first be
      # called with **kwargs, then the result of that call will be called
      # with the underlying function as the sole argument (i.e., it must be a
      # "decorator factory.").
      return (
          _batch_decorator(_func)
          if callable(_func)
          else _batch_decorator
      )
  
imports: import asyncio
  import time
  from dataclasses import dataclass
  from functools import wraps
  from inspect import isasyncgenfunction, iscoroutinefunction
  from typing import Any, AsyncGenerator, Callable, Dict, Iterable, List, Optional, Tuple, Type, TypeVar, overload
  from ray._private.signature import extract_signature, flatten_args, recover_args
  from ray._private.utils import get_or_create_event_loop
  from ray.serve._private.utils import extract_self_if_method_call
  from ray.serve.exceptions import RayServeException
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 6 for calling a global function with example usage: batch(_func, max_batch_size, batch_wait_timeout_s) and returns: unknown, wrapper, future, batch_handler_generator, unknown