docstr: Connect the Ray Client to a server.
  
  Args:
      conn_str: Connection string, in the form "[host]:port"
      job_config: The job config of the server.
      secure: Whether to use a TLS secured gRPC channel
      metadata: gRPC metadata to send on connect
      connection_retries: number of connection attempts to make
      ignore_version: whether to ignore Python or Ray version mismatches.
          This should only be used for debugging purposes.
  
  Returns:
      Dictionary of connection info, e.g., {"num_clients": 1}.
docstr_tok: ["Connect","the","Ray","Client","to","a","server",".","Args",":","conn_str",":","Connection","string",",","in","the","form","``","[","host","]",":","port","''","job_config",":","The","job","config","of","the","server",".","secure",":","Whether","to","use","a","TLS","secured","gRPC","channel","metadata",":","gRPC","metadata","to","send","on","connect","connection_retries",":","number","of","connection","attempts","to","make","ignore_version",":","whether","to","ignore","Python","or","Ray","version","mismatches",".","This","should","only","be","used","for","debugging","purposes",".","Returns",":","Dictionary","of","connection","info",",","e.g.",",","{","``","num_clients","''",":","1","}","."]
code: def connect(
      self,
      conn_str: str,
      job_config: JobConfig = None,
      secure: bool = False,
      metadata: List[Tuple[str, str]] = None,
      connection_retries: int = 3,
      namespace: str = None,
      *,
      ignore_version: bool = False,
      _credentials: Optional[
          "grpc.ChannelCredentials"
      ] = None,  # noqa: F821
      ray_init_kwargs: Optional[Dict[str, Any]] = None,
  ) -> Dict[str, Any]:
      """Connect the Ray Client to a server.
  
      Args:
          conn_str: Connection string, in the form "[host]:port"
          job_config: The job config of the server.
          secure: Whether to use a TLS secured gRPC channel
          metadata: gRPC metadata to send on connect
          connection_retries: number of connection attempts to make
          ignore_version: whether to ignore Python or Ray version mismatches.
              This should only be used for debugging purposes.
  
      Returns:
          Dictionary of connection info, e.g., {"num_clients": 1}.
      """
      # Delay imports until connect to avoid circular imports.
      from ray.util.client.worker import Worker
  
      if self.client_worker is not None:
          if self._connected_with_init:
              return
          raise Exception(
              "ray.init() called, but ray client is already connected"
          )
      if not self._inside_client_test:
          # If we're calling a client connect specifically and we're not
          # currently in client mode, ensure we are.
          _explicitly_enable_client_mode()
      if namespace is not None:
          job_config = job_config or JobConfig()
          job_config.set_ray_namespace(namespace)
  
      logging_level = ray_constants.LOGGER_LEVEL
      logging_format = ray_constants.LOGGER_FORMAT
  
      if ray_init_kwargs is None:
          ray_init_kwargs = {}
  
      # NOTE(architkulkarni): env_hook is not supported with Ray Client.
      ray_init_kwargs["_skip_env_hook"] = True
  
      if ray_init_kwargs.get("logging_level") is not None:
          logging_level = ray_init_kwargs["logging_level"]
      if ray_init_kwargs.get("logging_format") is not None:
          logging_format = ray_init_kwargs["logging_format"]
  
      setup_logger(logging_level, logging_format)
  
      try:
          self.client_worker = Worker(
              conn_str,
              secure=secure,
              _credentials=_credentials,
              metadata=metadata,
              connection_retries=connection_retries,
          )
          self.api.worker = self.client_worker
          self.client_worker._server_init(job_config, ray_init_kwargs)
          conn_info = self.client_worker.connection_info()
          self._check_versions(conn_info, ignore_version)
          self._register_serializers()
          return conn_info
      except Exception:
          self.disconnect()
          raise
  
imports: import logging
  import os
  import sys
  import threading
  from typing import Any, Dict, List, Optional, Tuple
  import ray._private.ray_constants
  from ray._private.client_mode_hook import _explicitly_disable_client_mode, _explicitly_enable_client_mode
  from ray._private.ray_logging import setup_logger
  from ray.job_config import JobConfig
  from ray.util.annotations import DeveloperAPI
label_desc: Use image node_id 2 for calling the _ClientContext obj's underlying member method code with example usage: obj.connect(conn_str, job_config, secure, metadata, connection_retries, namespace) and returns: conn_info