code: def create_serve_rest_api(
      dashboard_module_superclass: Union[
          dashboard_utils.DashboardHeadModule,
          dashboard_utils.DashboardAgentModule,
      ],
      dashboard_route_table: Union[
          dashboard_optional_utils.DashboardHeadRouteTable,
          dashboard_optional_utils.DashboardAgentRouteTable,
      ],
      log_deprecation_warning: bool = False,
  ):
      # NOTE (shrekris-anyscale): This class uses delayed imports for all
      # Ray Serve-related modules. That way, users can use the Ray dashboard agent for
      # non-Serve purposes without downloading Serve dependencies.
      class ServeRestApiImpl(dashboard_module_superclass):
          def __init__(self, dashboard_head_or_agent):
              super().__init__(dashboard_head_or_agent)
              self._controller = None
              self._controller_lock = asyncio.Lock()
  
              # serve_start_async is not thread-safe call. This lock
              # will make sure there is only one call that starts the serve instance.
              # If the lock is already acquired by another async task, the async task
              # will asynchronously wait for the lock.
              self._controller_start_lock = asyncio.Lock()
  
          # TODO: It's better to use `/api/version`.
          # It requires a refactor of ClassMethodRouteTable to differentiate the server.
          @dashboard_route_table.get("/api/ray/version")
          async def get_version(self, req: Request) -> Response:
              # NOTE(edoakes): CURRENT_VERSION should be bumped and checked on the
              # client when we have backwards-incompatible changes.
              resp = VersionResponse(
                  version=CURRENT_VERSION,
                  ray_version=ray.__version__,
                  ray_commit=ray.__commit__,
              )
              return Response(
                  text=json.dumps(dataclasses.asdict(resp)),
                  content_type="application/json",
                  status=aiohttp.web.HTTPOk.status_code,
              )
  
          @dashboard_route_table.get("/api/serve/applications/")
          @optional_utils.init_ray_and_catch_exceptions()
          @validate_endpoint(
              log_deprecation_warning=log_deprecation_warning
          )
          async def get_serve_instance_details(
              self, req: Request
          ) -> Response:
              from ray.serve.schema import ServeInstanceDetails
  
              controller = await self.get_serve_controller()
  
              if controller is None:
                  # If no serve instance is running, return a dict that represents that.
                  details = ServeInstanceDetails.get_empty_schema_dict()
              else:
                  try:
                      details = (
                          await controller.get_serve_instance_details.remote()
                      )
                  except ray.exceptions.RayTaskError as e:
                      # Task failure sometimes are due to GCS
                      # failure. When GCS failed, we expect a longer time
                      # to recover.
                      return Response(
                          status=503,
                          text=(
                              "Failed to get a response from the controller. "
                              f"The GCS may be down, please retry later: {e}"
                          ),
                      )
  
              return Response(
                  text=json.dumps(details),
                  content_type="application/json",
              )
  
          @dashboard_route_table.delete("/api/serve/applications/")
          @optional_utils.init_ray_and_catch_exceptions()
          async def delete_serve_applications(
              self, req: Request
          ) -> Response:
              from ray import serve
  
              if await self.get_serve_controller() is not None:
                  serve.shutdown()
  
              return Response()
  
          @dashboard_route_table.put("/api/serve/applications/")
          @optional_utils.init_ray_and_catch_exceptions()
          @validate_endpoint(
              log_deprecation_warning=log_deprecation_warning
          )
          async def put_all_applications(
              self, req: Request
          ) -> Response:
              from ray.serve.config import ProxyLocation
              from ray.serve._private.api import serve_start_async
              from ray.serve.schema import ServeDeploySchema
              from ray._private.usage.usage_lib import (
                  TagKey,
                  record_extra_usage_tag,
              )
  
              try:
                  config: ServeDeploySchema = (
                      ServeDeploySchema.parse_obj(await req.json())
                  )
              except ValidationError as e:
                  return Response(
                      status=400,
                      text=repr(e),
                  )
  
              config_http_options = config.http_options.dict()
              location = ProxyLocation._to_deployment_mode(
                  config.proxy_location
              )
              full_http_options = dict(
                  {"location": location}, **config_http_options
              )
              grpc_options = config.grpc_options.dict()
  
              async with self._controller_start_lock:
                  client = await serve_start_async(
                      http_options=full_http_options,
                      grpc_options=grpc_options,
                      global_logging_config=config.logging_config,
                  )
  
              # Serve ignores HTTP options if it was already running when
              # serve_start_async() is called. Therefore we validate that no
              # existing HTTP options are updated and print warning in case they are
              self.validate_http_options(client, full_http_options)
  
              try:
                  if config.logging_config:
                      client.update_global_logging_config(
                          config.logging_config
                      )
                  client.deploy_apps(config)
                  record_extra_usage_tag(
                      TagKey.SERVE_REST_API_VERSION, "v2"
                  )
              except RayTaskError as e:
                  return Response(
                      status=400,
                      text=str(e),
                  )
              else:
                  return Response()
  
          def validate_http_options(self, client, http_options):
              divergent_http_options = []
  
              for option, new_value in http_options.items():
                  prev_value = getattr(client.http_config, option)
                  if prev_value != new_value:
                      divergent_http_options.append(option)
  
              if divergent_http_options:
                  logger.warning(
                      "Serve is already running on this Ray cluster and "
                      "it's not possible to update its HTTP options without "
                      "restarting it. Following options are attempted to be "
                      f"updated: {divergent_http_options}."
                  )
  
          async def get_serve_controller(self):
              """Gets the ServeController to the this cluster's Serve app.
  
              return: If Serve is running on this Ray cluster, returns a client to
                  the Serve controller. If Serve is not running, returns None.
              """
              async with self._controller_lock:
                  if self._controller is not None:
                      try:
                          await self._controller.check_alive.remote()
                          return self._controller
                      except ray.exceptions.RayActorError:
                          logger.info("Controller is dead")
                      self._controller = None
  
                  # Try to connect to serve even when we detect the actor is dead
                  # because the user might have started a new
                  # serve cluter.
                  from ray.serve._private.constants import (
                      SERVE_CONTROLLER_NAME,
                      SERVE_NAMESPACE,
                  )
  
                  try:
                      # get_actor is a sync call but it'll timeout after
                      # ray.dashboard.consts.GCS_RPC_TIMEOUT_SECONDS
                      self._controller = ray.get_actor(
                          SERVE_CONTROLLER_NAME,
                          namespace=SERVE_NAMESPACE,
                      )
                  except Exception as e:
                      logger.debug(
                          "There is no "
                          "instance running on this Ray cluster. Please "
                          "call `serve.start(detached=True) to start "
                          f"one: {e}"
                      )
  
                  return self._controller
  
          async def run(self, server):
              pass
  
          @staticmethod
          def is_minimal_module():
              return False
  
      return ServeRestApiImpl(dashboard_module_superclass)
  
file_path: ray/dashboard/modules/serve/serve_rest_api_impl.py
imports: import json
  import asyncio
  import logging
  import dataclasses
  from functools import wraps
  from typing import Union
  import aiohttp
  from aiohttp.web import Request, Response
  import ray
  from ray.exceptions import RayTaskError
  from ray.dashboard.modules.version import CURRENT_VERSION, VersionResponse
  import ray.dashboard.utils
  import ray.dashboard.optional_utils
  import ray.dashboard.optional_utils
  from ray._private.pydantic_compat import ValidationError
label_desc: Use image node_id 2 for calling a global function with example usage: create_serve_rest_api(dashboard_module_superclass, dashboard_route_table, log_deprecation_warning) and returns: ServeRestApiImpl, Response, Response, Response, False, Response, self, Response, Response, Response, self