class_docstr: Attributes:
      name: run name
      ports: ports mapping, if run is attached
      backend: backend type
      status: run status
      hostname: instance hostname
class_docstr_tok: ["Attributes",":","name",":","run","name","ports",":","ports","mapping",",","if","run","is","attached","backend",":","backend","type","status",":","run","status","hostname",":","instance","hostname"]
docstr: Establish an SSH tunnel to the instance and update SSH config
  
  Args:
      ssh_identity_file: SSH keypair to access instances
  
  Raises:
      dstack.api.PortUsedError: If ports are in use or the run is attached by another process.
code: def attach(
      self,
      ssh_identity_file: Optional[PathLike] = None,
  ) -> bool:
      """
      Establish an SSH tunnel to the instance and update SSH config
  
      Args:
          ssh_identity_file: SSH keypair to access instances
  
      Raises:
          dstack.api.PortUsedError: If ports are in use or the run is attached by another process.
      """
      ssh_identity_file = ssh_identity_file or self._ssh_identity_file
      if ssh_identity_file is None:
          raise ConfigurationError(
              "SSH identity file is required to attach to the run"
          )
      ssh_identity_file = str(ssh_identity_file)
  
      if self._ssh_attach is None:
          while self.status in (
              RunStatus.SUBMITTED,
              RunStatus.PENDING,
              RunStatus.PROVISIONING,
              RunStatus.PULLING,
          ):
              time.sleep(5)
              self.refresh()
          # If status is done, there is a chance we could read logs from websocket
          if (
              self.status.is_finished()
              and self.status != RunStatus.DONE
          ):
              return False
  
          provisioning_data = (
              self._run.jobs[0]
              .job_submissions[-1]
              .job_provisioning_data
          )
  
          control_sock_path_and_port_locks = (
              SSHAttach.reuse_control_sock_path_and_port_locks(
                  run_name=self.name
              )
          )
  
          if control_sock_path_and_port_locks is None:
              if self._ports_lock is None:
                  self._ports_lock = _reserve_ports(
                      self._run.jobs[0].job_spec
                  )
  
              logger.debug(
                  "Attaching to %s (%s: %s)",
                  self.name,
                  provisioning_data.hostname,
                  self._ports_lock.dict(),
              )
          else:
              self._ports_lock = control_sock_path_and_port_locks[1]
  
              logger.debug(
                  "Reusing the existing tunnel to %s (%s: %s)",
                  self.name,
                  provisioning_data.hostname,
                  self._ports_lock.dict(),
              )
  
          self._ssh_attach = SSHAttach(
              hostname=self.hostname,
              ssh_port=provisioning_data.ssh_port,
              user=provisioning_data.username,
              id_rsa_path=ssh_identity_file,
              ports_lock=self._ports_lock,
              run_name=self.name,
              dockerized=provisioning_data.dockerized,
              control_sock_path=control_sock_path_and_port_locks[0]
              if control_sock_path_and_port_locks
              else None,
          )
          if not control_sock_path_and_port_locks:
              self._ssh_attach.attach()
          self._ports_lock = None
      return True
  
imports: import base64
  import queue
  import tempfile
  import threading
  import time
  from abc import ABC
  from copy import copy
  from datetime import datetime
  from pathlib import Path
  from typing import Dict, Iterable, List, Optional, Union
  import requests
  from websocket import WebSocketApp
  import dstack.api
  from dstack._internal.core.errors import ConfigurationError
  from dstack._internal.core.models.backends.base import BackendType
  from dstack._internal.core.models.configurations import AnyRunConfiguration
  from dstack._internal.core.models.profiles import Profile, ProfileResources, ProfileRetryPolicy, SpotPolicy
  from dstack._internal.core.models.repos.base import Repo
  from dstack._internal.core.models.runs import JobSpec
  from dstack._internal.core.models.runs import JobStatus
  from dstack._internal.core.models.runs import Run
  from dstack._internal.core.models.runs import RunPlan, RunSpec
  from dstack._internal.core.services.logs import URLReplacer
  from dstack._internal.core.services.ssh.attach import SSHAttach
  from dstack._internal.core.services.ssh.ports import PortsLock
  from dstack._internal.server.schemas.logs import PollLogsRequest
  from dstack._internal.utils.logging import get_logger
  from dstack._internal.utils.path import PathLike, path_in_dir
  from dstack.api.server import APIClient
label_desc: Use image node_id 12 for calling the Run obj's underlying member method code with example usage: obj.attach(ssh_identity_file) and returns: True, False