push/fetch: support targets in cloud version push/fetch

dvc/repo/fetch.py

@@ -53,8 +53,8 @@ def fetch(
 
     with suppress(NoRemoteError):
         _remote = self.cloud.get_remote(name=remote)
-        if _remote.worktree:
-            return fetch_worktree(self, _remote)
+        if _remote.worktree or _remote.fs.version_aware:
+            return fetch_worktree(self, _remote, targets=targets)
 
     failed_count = 0
 

dvc/repo/push.py

@@ -30,7 +30,7 @@ def push(
 
     _remote = self.cloud.get_remote(name=remote)
     if _remote.worktree or _remote.fs.version_aware:
-        return push_worktree(self, _remote)
+        return push_worktree(self, _remote, targets=targets)
 
     used_run_cache = (
         self.stage_cache.push(remote, odb=odb) if run_cache else []

dvc/repo/worktree.py

@@ -70,10 +70,12 @@ def worktree_view(
     )
 
 
-def fetch_worktree(repo: "Repo", remote: "Remote") -> int:
+def fetch_worktree(
+    repo: "Repo", remote: "Remote", targets: Optional["TargetType"] = None
+) -> int:
     from dvc_data.index import save
 
-    view = worktree_view(repo.index)
+    view = worktree_view(repo.index, targets=targets)
     index = view.data["repo"]
     for key, entry in index.iteritems():
         entry.fs = remote.fs
@@ -89,10 +91,14 @@ def fetch_worktree(repo: "Repo", remote: "Remote") -> int:
         return save(index, callback=cb)
 
 
-def push_worktree(repo: "Repo", remote: "Remote") -> int:
+def push_worktree(
+    repo: "Repo", remote: "Remote", targets: Optional["TargetType"] = None
+) -> int:
     from dvc_data.index import build, checkout
 
-    view = worktree_view(repo.index, push=True, latest_only=remote.worktree)
+    view = worktree_view(
+        repo.index, push=True, targets=targets, latest_only=remote.worktree
+    )
     new_index = view.data["repo"]
     if remote.worktree:
         logger.debug("indexing latest worktree for '%s'", remote.path)