code: def run(
      config_or_import_path: str,
      arguments: Tuple[str],
      runtime_env: str,
      runtime_env_json: str,
      working_dir: str,
      app_dir: str,
      address: str,
      host: str,
      port: int,
      blocking: bool,
      reload: bool,
      route_prefix: str,
      name: str,
  ):
      if host is not None or port is not None:
          cli_logger.warning(
              "Specifying `--host` and `--port` to `serve run` is deprecated and will be "
              "removed in a future version. To specify custom HTTP options, use the "
              "`serve start` command."
          )
  
      sys.path.insert(0, app_dir)
      args_dict = convert_args_to_dict(arguments)
      final_runtime_env = parse_runtime_env_args(
          runtime_env=runtime_env,
          runtime_env_json=runtime_env_json,
          working_dir=working_dir,
      )
  
      if pathlib.Path(config_or_import_path).is_file():
          if len(args_dict) > 0:
              cli_logger.warning(
                  "Application arguments are ignored when running a config file."
              )
  
          is_config = True
          config_path = config_or_import_path
          cli_logger.print(f"Running config file: '{config_path}'.")
  
          with open(config_path, "r") as config_file:
              config_dict = yaml.safe_load(config_file)
  
              config = ServeDeploySchema.parse_obj(config_dict)
  
              # If host or port is specified as a CLI argument, they should take
              # priority over config values.
              if host is None:
                  if "http_options" in config_dict:
                      host = config_dict["http_options"].get(
                          "host", DEFAULT_HTTP_HOST
                      )
                  else:
                      host = DEFAULT_HTTP_HOST
              if port is None:
                  if "http_options" in config_dict:
                      port = config_dict["http_options"].get(
                          "port", DEFAULT_HTTP_PORT
                      )
                  else:
                      port = DEFAULT_HTTP_PORT
  
      else:
          is_config = False
          if host is None:
              host = DEFAULT_HTTP_HOST
          if port is None:
              port = DEFAULT_HTTP_PORT
          import_path = config_or_import_path
          cli_logger.print(f"Running import path: '{import_path}'.")
          app = _private_api.call_app_builder_with_args_if_necessary(
              import_attr(import_path), args_dict
          )
  
      # Only initialize ray if it has not happened yet.
      if not ray.is_initialized():
          # Setting the runtime_env here will set defaults for the deployments.
          ray.init(
              address=address,
              namespace=SERVE_NAMESPACE,
              runtime_env=final_runtime_env,
          )
      elif (
          address is not None
          and address != "auto"
          and address != ray.get_runtime_context().gcs_address
      ):
          # Warning users the address they passed is different from the existing ray
          # instance.
          ray_address = ray.get_runtime_context().gcs_address
          cli_logger.warning(
              "An address was passed to `serve run` but the imported module also "
              f"connected to Ray at a different address: '{ray_address}'. You do not "
              "need to call `ray.init` in your code when using `serve run`."
          )
  
      http_options = {
          "host": host,
          "port": port,
          "location": "EveryNode",
      }
      grpc_options = gRPCOptions()
      # Merge http_options and grpc_options with the ones on ServeDeploySchema. If host
      # and/or port is passed by cli, those continue to take the priority
      if is_config and isinstance(config, ServeDeploySchema):
          config_http_options = config.http_options.dict()
          http_options = {**config_http_options, **http_options}
          grpc_options = gRPCOptions(**config.grpc_options.dict())
  
      client = _private_api.serve_start(
          http_options=http_options,
          grpc_options=grpc_options,
      )
  
      try:
          if is_config:
              client.deploy_apps(config, _blocking=False)
              cli_logger.success(
                  "Submitted deploy config successfully."
              )
          else:
              serve.run(
                  app,
                  name=name,
                  route_prefix=route_prefix,
                  host=host,
                  port=port,
              )
              cli_logger.success("Deployed Serve app successfully.")
  
          if reload:
              if not blocking:
                  raise click.ClickException(
                      "The --non-blocking option conflicts with the --reload option."
                  )
              if working_dir:
                  watch_dir = working_dir
              else:
                  watch_dir = app_dir
  
              for changes in watchfiles.watch(
                  watch_dir,
                  rust_timeout=10000,
                  yield_on_timeout=True,
              ):
                  if changes:
                      cli_logger.info(
                          f"Detected file change in path {watch_dir}. Redeploying app."
                      )
                      # The module needs to be reloaded with `importlib` in order to pick
                      # up any changes.
                      app = _private_api.call_app_builder_with_args_if_necessary(
                          import_attr(import_path, reload_module=True),
                          args_dict,
                      )
                      serve.run(
                          app,
                          name=name,
                          route_prefix=route_prefix,
                          host=host,
                          port=port,
                      )
  
          if blocking:
              while True:
                  # Block, letting Ray print logs to the terminal.
                  time.sleep(10)
  
      except KeyboardInterrupt:
          cli_logger.info("Got KeyboardInterrupt, shutting down...")
          serve.shutdown()
          sys.exit()
  
      except Exception:
          traceback.print_exc()
          cli_logger.error(
              "Received unexpected error, see console logs for more details. Shutting "
              "down..."
          )
          serve.shutdown()
          sys.exit()
  
imports: import os
  import pathlib
  import re
  import sys
  import time
  import traceback
  from dataclasses import asdict
  from typing import Dict, List, Optional, Tuple
  import click
  import watchfiles
  import yaml
  import ray
  from ray import serve
  from ray._private.utils import import_attr
  from ray.autoscaler._private.cli_logger import cli_logger
  from ray.dashboard.modules.dashboard_sdk import parse_runtime_env_args
  from ray.dashboard.modules.serve.sdk import ServeSubmissionClient
  from ray.serve._private import api
  from ray.serve._private.constants import DEFAULT_GRPC_PORT, DEFAULT_HTTP_HOST, DEFAULT_HTTP_PORT, SERVE_DEFAULT_APP_NAME, SERVE_NAMESPACE
  from ray.serve._private.deployment_graph_build import build
  from ray.serve._private.deployment_graph_build import get_and_validate_ingress_deployment
  from ray.serve.config import DeploymentMode, ProxyLocation, gRPCOptions
  from ray.serve.deployment import Application, deployment_to_schema
  from ray.serve.schema import LoggingConfig, ServeApplicationSchema, ServeDeploySchema, ServeInstanceDetails
label_desc: Use image node_id 9 for calling a global function with example usage: run(config_or_import_path, arguments, runtime_env, runtime_env_json, working_dir, app_dir, address, host, port, blocking, reload, route_prefix, name) without return types