exp: add contextmanager for switching to active tempdir repo

(cherry picked from commit 8cb2646023)

dvc/repo/experiments/brancher.py

@@ -0,0 +1,43 @@
+from contextlib import ExitStack, contextmanager
+from typing import TYPE_CHECKING, Iterator, Tuple
+
+from dvc.repo.experiments.exceptions import InvalidExpRevError
+from dvc.scm import RevError
+
+if TYPE_CHECKING:
+    from dvc.repo import Repo
+
+
+@contextmanager
+def switch_repo(
+    repo: "Repo",
+    rev: str,
+) -> Iterator[Tuple["Repo", str]]:
+    """Return a repo instance (brancher) switched to rev.
+
+    If rev is the name of a running experiment, the returned instance will be
+    the live repo wherever the experiment is running.
+
+    NOTE: This will not resolve git SHA's that only exist in queued exp workspaces
+    (it will only match queued exp names).
+    """
+    try:
+        with repo.switch(rev):
+            yield repo, rev
+        return
+    except RevError as exc:
+        orig_exc = exc
+    exps = repo.experiments
+
+    for queue in (exps.tempdir_queue, exps.celery_queue):
+        try:
+            active_repo = queue.active_repo(rev)
+        except InvalidExpRevError:
+            continue
+        stack = ExitStack()
+        stack.enter_context(active_repo)
+        stack.enter_context(active_repo.switch("workspace"))
+        with stack:
+            yield active_repo, rev
+        return
+    raise orig_exc

dvc/repo/experiments/exceptions.py

@@ -114,3 +114,10 @@ class UnresolvedRunningExpNamesError(UnresolvedExpNamesError):
 
 class ExpQueueEmptyError(DvcException):
     pass
+
+
+class ExpNotStartedError(DvcException):
+    def __init__(self, name: str):
+        super().__init__(
+            f"Queued experiment '{name}' exists but has not started running yet"
+        )

dvc/repo/experiments/queue/base.py

@@ -790,3 +790,26 @@ class BaseStashQueue(ABC):
         Returns:
             Dict mapping baseline revision to list of queued experiments.
         """
+
+    def active_repo(self, name: str) -> "Repo":
+        """Return a Repo for the specified active experiment if it exists."""
+        from dvc.repo import Repo
+        from dvc.repo.experiments.exceptions import (
+            ExpNotStartedError,
+            InvalidExpRevError,
+        )
+        from dvc.repo.experiments.executor.base import ExecutorInfo, TaskStatus
+
+        for entry in self.iter_active():
+            if entry.name != name:
+                continue
+            infofile = self.get_infofile_path(entry.stash_rev)
+            executor_info = ExecutorInfo.load_json(infofile)
+            if executor_info.status < TaskStatus.RUNNING:
+                raise ExpNotStartedError(name)
+            dvc_root = os.path.join(executor_info.root_dir, executor_info.dvc_dir)
+            try:
+                return Repo(dvc_root)
+            except (FileNotFoundError, DvcException) as exc:
+                raise InvalidExpRevError(name) from exc
+        raise InvalidExpRevError(name)