try force overwriting run cache entries on push

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

.pre-commit-config.yaml

@@ -22,13 +22,13 @@ repos:
       - id: sort-simple-yaml
       - id: trailing-whitespace
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.4.1'
+    rev: 'v0.5.4'
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]
       - id: ruff-format
   - repo: https://github.com/codespell-project/codespell
-    rev: v2.2.6
+    rev: v2.3.0
     hooks:
       - id: codespell
         additional_dependencies: ["tomli"]

dvc/api/data.py

@@ -285,7 +285,7 @@ def _open(
             if TYPE_CHECKING:
                 from dvc.fs import FileSystem
 
-            fs: Union["FileSystem", DataFileSystem, DVCFileSystem]
+            fs: Union[FileSystem, DataFileSystem, DVCFileSystem]
             if os.path.isabs(path):
                 fs = DataFileSystem(index=_repo.index.data["local"])
                 fs_path = path

dvc/cli/command.py

@@ -18,8 +18,8 @@ class CmdBase(ABC):
 
         os.chdir(args.cd)
 
-        self.repo: "Repo" = Repo(uninitialized=self.UNINITIALIZED)
-        self.config: "Config" = self.repo.config
+        self.repo: Repo = Repo(uninitialized=self.UNINITIALIZED)
+        self.config: Config = self.repo.config
         self.args = args
 
     def do_run(self):

dvc/commands/artifacts.py

@@ -13,6 +13,9 @@ class CmdArtifactsGet(CmdBaseNoRepo):
         from dvc.scm import CloneError
         from dvc.ui import ui
 
+        if self.args.show_url:
+            return self._show_url()
+
         try:
             count, out = Artifacts.get(
                 self.args.url,
@@ -36,6 +39,28 @@ class CmdArtifactsGet(CmdBaseNoRepo):
             )
             return 1
 
+    def _show_url(self):
+        from dvc.api import artifacts_show, get_url
+        from dvc.ui import ui
+
+        artifact = artifacts_show(
+            self.args.name,
+            version=self.args.rev,
+            stage=self.args.stage,
+            repo=self.args.url,
+        )
+
+        url = get_url(
+            artifact["path"],
+            repo=self.args.url,
+            rev=artifact["rev"],
+            remote=self.args.remote,
+            remote_config=self.args.remote_config,
+        )
+        ui.write(url, force=True)
+
+        return 0
+
 
 def add_parser(subparsers, parent_parser):
     ARTIFACTS_HELP = "DVC model registry artifact commands."
@@ -84,6 +109,14 @@ def add_parser(subparsers, parent_parser):
         help="Destination path to download artifact to",
         metavar="<path>",
     ).complete = completion.DIR
+    get_parser.add_argument(
+        "--show-url",
+        action="store_true",
+        help=(
+            "Print the storage location (URL) the target data would be "
+            "downloaded from, and exit."
+        ),
+    )
     get_parser.add_argument(
         "-j",
         "--jobs",

dvc/commands/data_sync.py

@@ -194,8 +194,8 @@ def add_parser(subparsers, _parent_parser):
     )
     pull_parser.add_argument(
         "--run-cache",
-        action="store_true",
-        default=False,
+        action=argparse.BooleanOptionalAction,
+        default=True,
         help="Fetch run history for all stages.",
     )
     pull_parser.add_argument(
@@ -262,8 +262,8 @@ def add_parser(subparsers, _parent_parser):
     )
     push_parser.add_argument(
         "--run-cache",
-        action="store_true",
-        default=False,
+        action=argparse.BooleanOptionalAction,
+        default=True,
         help="Push run history for all stages.",
     )
     push_parser.add_argument(
@@ -324,8 +324,8 @@ def add_parser(subparsers, _parent_parser):
     )
     fetch_parser.add_argument(
         "--run-cache",
-        action="store_true",
-        default=False,
+        action=argparse.BooleanOptionalAction,
+        default=True,
         help="Fetch run history for all stages.",
     )
     fetch_parser.add_argument(

dvc/commands/dataset.py

@@ -42,7 +42,7 @@ class CmdDatasetAdd(CmdBase):
             if rev := dataset.lock.rev:
                 ver = rev
 
-        ver_part: Optional["Text"] = None
+        ver_part: Optional[Text] = None
         if ver:
             ver_part = ui.rich_text.assemble(" @ ", (ver, "repr.number"))
         text = ui.rich_text.assemble("(", (url, "repr.url"), ver_part or "", ")")

dvc/commands/experiments/pull.py

@@ -1,3 +1,5 @@
+import argparse
+
 from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
@@ -81,8 +83,8 @@ def add_parser(experiments_subparsers, parent_parser):
     )
     experiments_pull_parser.add_argument(
         "--run-cache",
-        action="store_true",
-        default=False,
+        action=argparse.BooleanOptionalAction,
+        default=True,
         help="Pull run history for all stages.",
     )
     experiments_pull_parser.add_argument(

dvc/commands/experiments/push.py

@@ -1,3 +1,4 @@
+import argparse
 from typing import Any
 
 from dvc.cli import completion, formatter
@@ -120,8 +121,8 @@ def add_parser(experiments_subparsers, parent_parser):
     )
     experiments_push_parser.add_argument(
         "--run-cache",
-        action="store_true",
-        default=False,
+        action=argparse.BooleanOptionalAction,
+        default=True,
         help="Push run history for all stages.",
     )
     experiments_push_parser.add_argument(

dvc/commands/gc.py

@@ -60,7 +60,7 @@ class CmdGC(CmdBase):
             msg += " of the current and the following repos:"
 
             for repo_path in self.args.repos:
-                msg += "\n  - %s" % os.path.abspath(repo_path)
+                msg += f"\n  - {os.path.abspath(repo_path)}"
         else:
             msg += " of the current repo."
 

dvc/commands/plots.py

@@ -72,7 +72,7 @@ class CmdPlots(CmdBase):
                 html_template_path = os.path.join(self.repo.dvc_dir, html_template_path)
         return html_template_path
 
-    def run(self) -> int:  # noqa: C901, PLR0911, PLR0912
+    def run(self) -> int:  # noqa: C901, PLR0911
         from pathlib import Path
 
         from dvc.render.match import match_defs_renderers

dvc/commands/repro.py

@@ -117,10 +117,7 @@ and then the stage name name.
         "--pull",
         action="store_true",
         default=False,
-        help=(
-            "Try automatically pulling missing cache for outputs restored "
-            "from the run-cache."
-        ),
+        help="Try automatically pulling missing data.",
     )
     repro_parser.add_argument(
         "--allow-missing",

dvc/commands/stage.py

@@ -68,7 +68,7 @@ def prepare_stages_data(
 class CmdStageList(CmdBase):
     def _get_stages(self) -> Iterable["Stage"]:
         if self.args.all:
-            stages: list["Stage"] = self.repo.index.stages
+            stages: list[Stage] = self.repo.index.stages
             logger.trace("%d no. of stages found", len(stages))
             return stages
 

dvc/commands/studio.py

@@ -18,11 +18,16 @@ class CmdStudioLogin(CmdConfig):
         from dvc.utils.studio import STUDIO_URL
         from dvc_studio_client.auth import StudioAuthError, get_access_token
 
+        studio = self.config.get("studio", {})
         name = self.args.name
-        hostname = self.args.hostname or os.environ.get(DVC_STUDIO_URL) or STUDIO_URL
+        hostname = (
+            self.args.hostname
+            or os.environ.get(DVC_STUDIO_URL)
+            or studio.get("url")
+            or STUDIO_URL
+        )
         scopes = self.args.scopes
 
-        studio = self.config.get("studio", {})
         if studio.get("url", hostname) == hostname and "token" in studio:
             raise DvcException(
                 "Token already exists. "

dvc/compare.py

@@ -160,7 +160,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             if key not in keys:
                 self.add_column(key)
 
-        row: list["CellT"] = [
+        row: list[CellT] = [
             with_value(d.get(key), self._fill_value) for key in self.keys()
         ]
         self.append(row)

dvc/daemon.py

@@ -44,10 +44,10 @@ def _win_detached_subprocess(args: Sequence[str], **kwargs) -> int:
 
     startupinfo = STARTUPINFO()
     startupinfo.dwFlags |= STARTF_USESHOWWINDOW
-    popen = subprocess.Popen(
+    popen = subprocess.Popen(  # noqa: S603
         args,
         close_fds=True,
-        shell=False,  # noqa: S603
+        shell=False,
         startupinfo=startupinfo,
         creationflags=creationflags,
         **kwargs,
@@ -149,7 +149,7 @@ def _spawn(
     env: Optional[Mapping[str, str]] = None,
     output_file: Optional[str] = None,
 ) -> int:
-    file: "AbstractContextManager[Any]" = nullcontext()
+    file: AbstractContextManager[Any] = nullcontext()
     kwargs = {}
     if output_file:
         file = open(output_file, "ab")  # noqa: SIM115

dvc/dependency/db.py

@@ -37,7 +37,7 @@ class AbstractDependency(Dependency):
     """Dependency without workspace/fs/fs_path"""
 
     def __init__(self, stage: "Stage", info: dict[str, Any], *args, **kwargs):
-        self.repo: "Repo" = stage.repo
+        self.repo: Repo = stage.repo
         self.stage = stage
         self.fs = None
         self.fs_path = None

dvc/dependency/repo.py

@@ -35,7 +35,7 @@ class RepoDependency(Dependency):
         super().__init__(stage, *args, **kwargs)
 
         self.fs = self._make_fs()
-        self.fs_path = as_posix(self.def_path)
+        self.fs_path = as_posix(self.fs.normpath(self.def_path))
 
     def _parse_path(self, fs, fs_path):  # noqa: ARG002
         return None

dvc/fs/__init__.py

@@ -1,3 +1,4 @@
+import glob
 from typing import Optional
 from urllib.parse import urlparse
 
@@ -69,7 +70,14 @@ def download(
             to_infos = [to]
 
         if isinstance(fs, DVCFileSystem):
-            lfs_prefetch(fs, from_infos)
+            lfs_prefetch(
+                fs,
+                [
+                    f"{fs.normpath(glob.escape(fs_path))}/**"
+                    if fs.isdir(fs_path)
+                    else glob.escape(fs_path)
+                ],
+            )
         cb.set_size(len(from_infos))
         jobs = jobs or fs.jobs
         generic.copy(fs, from_infos, localfs, to_infos, callback=cb, batch_size=jobs)

dvc/fs/dvc.py

@@ -360,7 +360,7 @@ class _DVCFileSystem(AbstractFileSystem):
         except FileNotFoundError:
             return False
 
-    def ls(self, path, detail=True, dvc_only=False, **kwargs):  # noqa: C901
+    def ls(self, path, detail=True, dvc_only=False, **kwargs):  # noqa: C901, PLR0912
         key = self._get_key_from_relative(path)
         repo, dvc_fs, subkey = self._get_subrepo_info(key)
 
@@ -429,7 +429,7 @@ class _DVCFileSystem(AbstractFileSystem):
         ignore_subrepos = kwargs.get("ignore_subrepos", True)
         return self._info(key, path, ignore_subrepos=ignore_subrepos)
 
-    def _info(  # noqa: C901, PLR0912
+    def _info(  # noqa: C901
         self, key, path, ignore_subrepos=True, check_ignored=True
     ):
         repo, dvc_fs, subkey = self._get_subrepo_info(key)

dvc/output.py

@@ -390,7 +390,7 @@ class Output:
         self.can_push = push
 
         self.fs_path = self._parse_path(self.fs, fs_path)
-        self.obj: Optional["HashFile"] = None
+        self.obj: Optional[HashFile] = None
 
         self.remote = remote
 
@@ -902,7 +902,7 @@ class Output:
     def get_obj(
         self, filter_info: Optional[str] = None, **kwargs
     ) -> Optional["HashFile"]:
-        obj: Optional["HashFile"] = None
+        obj: Optional[HashFile] = None
         if self.obj:
             obj = self.obj
         elif self.files:
@@ -1158,7 +1158,7 @@ class Output:
             logger.warning(msg)
             return {}
 
-        obj: Optional["HashFile"]
+        obj: Optional[HashFile]
         if self.is_dir_checksum:
             obj = self._collect_used_dir_cache(**kwargs)
         else:
@@ -1353,7 +1353,7 @@ class Output:
         cache = self.cache if self.use_cache else self.local_cache
         assert isinstance(cache, HashFileDB)
 
-        new: "HashFile"
+        new: HashFile
         try:
             assert self.hash_name
             staging, meta, obj = self._build(

dvc/parsing/__init__.py

@@ -563,10 +563,10 @@ class MatrixDefinition:
         iterable = self.resolved_iterable
         assert isinstance(iterable, Mapping)
 
-        ret: dict[str, "DictStrAny"] = {}
+        ret: dict[str, DictStrAny] = {}
         matrix = {key: enumerate(v) for key, v in iterable.items()}
         for combination in product(*matrix.values()):
-            d: "DictStrAny" = {}
+            d: DictStrAny = {}
             fragments: list[str] = []
             for k, (i, v) in zip(matrix.keys(), combination):
                 d[k] = v

dvc/progress.py

@@ -41,7 +41,7 @@ class Tqdm(tqdm):
         "miniters": 1,
     }
 
-    def __init__(  # noqa: PLR0913
+    def __init__(
         self,
         iterable=None,
         disable=None,

dvc/repo/__init__.py

@@ -184,12 +184,12 @@ class Repo:
         # used by DVCFileSystem to determine if it should traverse subrepos
         self.subrepos = subrepos
 
-        self.cloud: "DataCloud" = DataCloud(self)
-        self.stage: "StageLoad" = StageLoad(self)
+        self.cloud: DataCloud = DataCloud(self)
+        self.stage: StageLoad = StageLoad(self)
 
-        self.lock: "LockBase"
+        self.lock: LockBase
         self.cache: CacheManager
-        self.state: "StateBase"
+        self.state: StateBase
         if isinstance(self.fs, GitFileSystem) or not self.dvc_dir:
             self.lock = LockNoop()
             self.state = StateNoop()

dvc/repo/artifacts.py

@@ -137,14 +137,12 @@ class Artifacts:
 
         assert not (version and stage)
         name = _reformat_name(name)
-        tags: list["GitTag"] = find(
-            name=name, version=version, stage=stage, scm=self.scm
-        )
+        tags: list[GitTag] = find(name=name, version=version, stage=stage, scm=self.scm)
         if not tags:
             raise ArtifactNotFoundError(name, version=version, stage=stage)
         if version or stage:
             return tags[-1].target
-        gto_tags: list["GTOTag"] = sort_versions(parse_tag(tag) for tag in tags)
+        gto_tags: list[GTOTag] = sort_versions(parse_tag(tag) for tag in tags)
         return gto_tags[0].tag.target
 
     @classmethod
@@ -267,7 +265,7 @@ class Artifacts:
                 from_infos.append(url)
         except DvcException:
             raise
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             raise DvcException(
                 f"Failed to download artifact '{name}' via Studio"
             ) from exc
@@ -285,7 +283,7 @@ class Artifacts:
         return len(to_infos), relpath(localfs.commonpath(to_infos))
 
     @classmethod
-    def get(  # noqa: PLR0913
+    def get(
         cls,
         url: str,
         name: str,
@@ -314,6 +312,13 @@ class Artifacts:
 
         name = _reformat_name(name)
         saved_exc: Optional[Exception] = None
+
+        local_dvc_studio_config = Config().get("studio", {})
+        args_dvc_studio_config = {}
+        if config and not isinstance(config, dict):
+            config = Config.load_file(config)
+            args_dvc_studio_config = config.get("studio", {})
+
         try:
             logger.trace("Trying studio-only config")
             return cls._download_studio(
@@ -324,14 +329,13 @@ class Artifacts:
                 out=out,
                 force=force,
                 jobs=jobs,
+                dvc_studio_config=local_dvc_studio_config | args_dvc_studio_config,
             )
         except FileExistsLocallyError:
             raise
         except Exception as exc:  # noqa: BLE001
             saved_exc = exc
 
-        if config and not isinstance(config, dict):
-            config = Config.load_file(config)
         with Repo.open(
             url=url,
             subrepos=True,
@@ -341,7 +345,7 @@ class Artifacts:
             remote_config=remote_config,
         ) as repo:
             logger.trace("Trying repo [studio] config")
-            dvc_studio_config = dict(repo.config.get("studio"))
+            repo_dvc_studio_config = repo.config.get("studio", {})
             try:
                 return cls._download_studio(
                     url,
@@ -351,7 +355,9 @@ class Artifacts:
                     out=out,
                     force=force,
                     jobs=jobs,
-                    dvc_studio_config=dvc_studio_config,
+                    dvc_studio_config=local_dvc_studio_config
+                    | repo_dvc_studio_config
+                    | args_dvc_studio_config,
                 )
             except FileExistsLocallyError:
                 raise
@@ -369,7 +375,7 @@ class Artifacts:
                 )
             except FileExistsLocallyError:
                 raise
-            except Exception as exc:  # noqa: BLE001
+            except Exception as exc:
                 if saved_exc:
                     logger.exception(str(saved_exc), exc_info=saved_exc.__cause__)
                 raise DvcException(

dvc/repo/checkout.py

@@ -125,7 +125,7 @@ def checkout(  # noqa: C901
             (StageFileDoesNotExistError, StageFileBadNameError, NoOutputOrStageError),
         ):
             raise CheckoutErrorSuggestGit(target) from exc
-        raise
+        raise  # noqa: PLE0704
 
     view = self.index.targets_view(
         targets, recursive=recursive, with_deps=with_deps, onerror=onerror

dvc/repo/datasets.py

@@ -242,7 +242,7 @@ class DatasetNotFoundError(DvcException, KeyError):
 
 class Datasets(Mapping[str, Dataset]):
     def __init__(self, repo: "Repo") -> None:
-        self.repo: "Repo" = repo
+        self.repo: Repo = repo
 
     def __repr__(self):
         return repr(dict(self))

dvc/repo/diff.py

@@ -26,7 +26,7 @@ def _diff(old, new, data_keys, with_missing=False):
     from dvc_data.index.diff import ADD, DELETE, MODIFY, RENAME
     from dvc_data.index.diff import diff as idiff
 
-    ret: "dict[str, list[dict]]" = {
+    ret: dict[str, list[dict]] = {
         "added": [],
         "deleted": [],
         "modified": [],

dvc/repo/experiments/__init__.py

@@ -119,7 +119,7 @@ class Experiments:
         **kwargs,
     ):
         """Reproduce and checkout a single (standalone) experiment."""
-        exp_queue: "BaseStashQueue" = (
+        exp_queue: BaseStashQueue = (
             self.tempdir_queue if tmp_dir else self.workspace_queue
         )
         self.queue_one(exp_queue, **kwargs)

dvc/repo/experiments/apply.py

@@ -25,7 +25,7 @@ def apply(repo: "Repo", rev: str, **kwargs):
     from dvc.repo.checkout import checkout as dvc_checkout
     from dvc.scm import RevError, resolve_rev
 
-    exps: "Experiments" = repo.experiments
+    exps: Experiments = repo.experiments
 
     is_stash: bool = False
 
@@ -57,7 +57,7 @@ def apply(repo: "Repo", rev: str, **kwargs):
 
 
 def _apply(repo: "Repo", rev: str, name: Optional[str] = None, is_stash: bool = False):
-    exps: "Experiments" = repo.experiments
+    exps: Experiments = repo.experiments
 
     with exps.apply_stash.preserve_workspace(rev, name=name):
         with repo.scm.detach_head(rev, force=True):

dvc/repo/experiments/collect.py

@@ -69,11 +69,11 @@ def collect_rev(
 
 def _collect_rev(
     repo: "Repo",
-    rev: str,
+    revision: str,
     param_deps: bool = False,
     **kwargs,
 ) -> SerializableExp:
-    with repo.switch(rev) as rev:
+    with repo.switch(revision) as rev:
         if rev == "workspace":
             timestamp: Optional[datetime] = None
         else:
@@ -174,7 +174,7 @@ def collect_active(
     """
     if not baseline_revs:
         return {}
-    result: dict[str, list["ExpRange"]] = {}
+    result: dict[str, list[ExpRange]] = {}
     exps = repo.experiments
     for queue in (exps.workspace_queue, exps.tempdir_queue, exps.celery_queue):
         for baseline, active_exps in queue.collect_active_data(
@@ -220,7 +220,7 @@ def collect_successful(
     Returns:
         Dict mapping baseline revision to successful experiments.
     """
-    result: dict[str, list["ExpRange"]] = {}
+    result: dict[str, list[ExpRange]] = {}
     for baseline_rev in baseline_revs:
         result[baseline_rev] = list(_collect_baseline(repo, baseline_rev, **kwargs))
     return result
@@ -304,7 +304,7 @@ def collect(
         )
 
     workspace_data = collect_rev(repo, "workspace", **kwargs)
-    result: list["ExpState"] = [workspace_data]
+    result: list[ExpState] = [workspace_data]
     queued = collect_queued(repo, baseline_revs, **kwargs) if not hide_queued else {}
     active = collect_active(repo, baseline_revs, **kwargs)
     failed = collect_failed(repo, baseline_revs, **kwargs) if not hide_failed else {}

dvc/repo/experiments/executor/base.py

@@ -199,7 +199,7 @@ class BaseExecutor(ABC):
     @classmethod
     def from_info(cls, info: "ExecutorInfo") -> "Self":
         if info.result_hash:
-            result: Optional["ExecutorResult"] = ExecutorResult(
+            result: Optional[ExecutorResult] = ExecutorResult(
                 info.result_hash,
                 (ExpRefInfo.from_ref(info.result_ref) if info.result_ref else None),
                 info.result_force,
@@ -470,7 +470,7 @@ class BaseExecutor(ABC):
             cls._set_log_level(log_level)
 
         exp_hash: Optional[str] = None
-        exp_ref: Optional["ExpRefInfo"] = None
+        exp_ref: Optional[ExpRefInfo] = None
         repro_force: bool = False
 
         if info.name:
@@ -559,7 +559,7 @@ class BaseExecutor(ABC):
             )
 
         ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
-        exp_ref: Optional["ExpRefInfo"] = ExpRefInfo.from_ref(ref) if ref else None
+        exp_ref: Optional[ExpRefInfo] = ExpRefInfo.from_ref(ref) if ref else None
         if cls.WARN_UNTRACKED:
             untracked = dvc.scm.untracked_files()
             if untracked:

dvc/repo/experiments/executor/local.py

@@ -201,7 +201,7 @@ class WorkspaceExecutor(BaseLocalExecutor):
     @classmethod
     def from_stash_entry(cls, repo: "Repo", entry: "ExpStashEntry", **kwargs):
         root_dir = repo.scm.root_dir
-        executor: "WorkspaceExecutor" = cls._from_stash_entry(
+        executor: WorkspaceExecutor = cls._from_stash_entry(
             repo, entry, root_dir, **kwargs
         )
         logger.debug("Init workspace executor in '%s'", root_dir)

dvc/repo/experiments/pull.py

@@ -30,7 +30,7 @@ def pull(  # noqa: C901
     pull_cache: bool = False,
     **kwargs,
 ) -> Iterable[str]:
-    exp_ref_set: set["ExpRefInfo"] = set()
+    exp_ref_set: set[ExpRefInfo] = set()
     if all_commits:
         exp_ref_set.update(exp_refs(repo.scm, git_remote))
     elif exp_names:
@@ -97,7 +97,7 @@ def _pull(
     def group_result(refspec):
         return results[str(refspec)]
 
-    pull_result: Mapping[SyncStatus, list["ExpRefInfo"]] = group_by(group_result, refs)
+    pull_result: Mapping[SyncStatus, list[ExpRefInfo]] = group_by(group_result, refs)
 
     return pull_result
 

dvc/repo/experiments/push.py

@@ -96,7 +96,7 @@ def push(
     push_cache: bool = False,
     **kwargs: Any,
 ) -> dict[str, Any]:
-    exp_ref_set: set["ExpRefInfo"] = set()
+    exp_ref_set: set[ExpRefInfo] = set()
     assert isinstance(repo.scm, Git)
     if all_commits:
         exp_ref_set.update(exp_refs(repo.scm))
@@ -162,7 +162,7 @@ def _push(
     def group_result(refspec):
         return results[str(refspec)]
 
-    pull_result: Mapping[SyncStatus, list["ExpRefInfo"]] = group_by(group_result, refs)
+    pull_result: Mapping[SyncStatus, list[ExpRefInfo]] = group_by(group_result, refs)
 
     return pull_result
 

dvc/repo/experiments/queue/celery.py

@@ -470,11 +470,11 @@ class LocalCeleryQueue(BaseStashQueue):
             exp_names = [exp_names]
         results: dict[str, ExpRefAndQueueEntry] = {}
 
-        exp_ref_match: dict[str, Optional["ExpRefInfo"]] = resolve_name(
+        exp_ref_match: dict[str, Optional[ExpRefInfo]] = resolve_name(
             self.scm, exp_names, git_remote
         )
         if not git_remote:
-            queue_entry_match: dict[str, Optional["QueueEntry"]] = (
+            queue_entry_match: dict[str, Optional[QueueEntry]] = (
                 self.match_queue_entry_by_name(
                     exp_names, self.iter_queued(), self.iter_done()
                 )
@@ -613,7 +613,7 @@ class LocalCeleryQueue(BaseStashQueue):
         """Map exp refs to any available successful executors."""
         from dvc.repo.experiments.serialize import ExpExecutor, LocalExpExecutor
 
-        result: dict[str, "ExpExecutor"] = {}
+        result: dict[str, ExpExecutor] = {}
         for entry, exec_result in self.iter_success():
             if baseline_revs and entry.baseline_rev not in baseline_revs:
                 continue

dvc/repo/experiments/queue/remove.py

@@ -21,11 +21,11 @@ def remove_tasks(  # noqa: C901, PLR0912
     """
     from celery.result import AsyncResult
 
-    stash_revs: dict[str, "ExpStashEntry"] = {}
-    failed_stash_revs: list["ExpStashEntry"] = []
-    done_entry_set: set["QueueEntry"] = set()
+    stash_revs: dict[str, ExpStashEntry] = {}
+    failed_stash_revs: list[ExpStashEntry] = []
+    done_entry_set: set[QueueEntry] = set()
     stash_rev_all = celery_queue.stash.stash_revs
-    failed_rev_all: dict[str, "ExpStashEntry"] = {}
+    failed_rev_all: dict[str, ExpStashEntry] = {}
     if celery_queue.failed_stash:
         failed_rev_all = celery_queue.failed_stash.stash_revs
     for entry in queue_entries:
@@ -90,17 +90,17 @@ def celery_clear(
     """
 
     removed: list[str] = []
-    entry_list: list["QueueEntry"] = []
+    entry_list: list[QueueEntry] = []
     if queued:
-        queue_entries: list["QueueEntry"] = list(self.iter_queued())
+        queue_entries: list[QueueEntry] = list(self.iter_queued())
         entry_list.extend(queue_entries)
         removed.extend(_get_names(queue_entries))
     if failed:
-        failed_tasks: list["QueueDoneResult"] = list(self.iter_failed())
+        failed_tasks: list[QueueDoneResult] = list(self.iter_failed())
         entry_list.extend([result.entry for result in failed_tasks])
         removed.extend(_get_names(failed_tasks))
     if success:
-        success_tasks: list["QueueDoneResult"] = list(self.iter_success())
+        success_tasks: list[QueueDoneResult] = list(self.iter_success())
         entry_list.extend([result.entry for result in success_tasks])
         removed.extend(_get_names(success_tasks))
 
@@ -125,7 +125,7 @@ def celery_remove(self: "LocalCeleryQueue", revs: Collection[str]) -> list[str]:
 
     remained: list[str] = []
     removed: list[str] = []
-    entry_to_remove: list["QueueEntry"] = []
+    entry_to_remove: list[QueueEntry] = []
     for name, entry in match_results.items():
         if entry:
             entry_to_remove.append(entry)

dvc/repo/experiments/queue/tempdir.py

@@ -125,7 +125,7 @@ class TempDirQueue(WorkspaceQueue):
                 )
         except DvcException:
             raise
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'") from exc
         finally:
             executor.cleanup(infofile)

dvc/repo/experiments/queue/workspace.py

@@ -131,7 +131,7 @@ class WorkspaceQueue(BaseStashQueue):
                 )
         except DvcException:
             raise
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'") from exc
         finally:
             executor.cleanup(infofile)

dvc/repo/experiments/remove.py

@@ -34,17 +34,17 @@ def remove(  # noqa: C901, PLR0912
     removed: list[str] = []
     if not any([exp_names, queue, all_commits, rev]):
         return removed
-    celery_queue: "LocalCeleryQueue" = repo.experiments.celery_queue
+    celery_queue: LocalCeleryQueue = repo.experiments.celery_queue
 
     if queue:
         removed.extend(celery_queue.clear(queued=True))
 
     assert isinstance(repo.scm, Git)
 
-    exp_ref_list: list["ExpRefInfo"] = []
-    queue_entry_list: list["QueueEntry"] = []
+    exp_ref_list: list[ExpRefInfo] = []
+    queue_entry_list: list[QueueEntry] = []
     if exp_names:
-        results: dict[str, "ExpRefAndQueueEntry"] = (
+        results: dict[str, ExpRefAndQueueEntry] = (
             celery_queue.get_ref_and_entry_by_names(exp_names, git_remote)
         )
         remained: list[str] = []
@@ -94,7 +94,7 @@ def _resolve_exp_by_baseline(
 ) -> dict[str, "ExpRefInfo"]:
     assert isinstance(repo.scm, Git)
 
-    commit_ref_dict: dict[str, "ExpRefInfo"] = {}
+    commit_ref_dict: dict[str, ExpRefInfo] = {}
     rev_dict = iter_revs(repo.scm, rev, num)
     rev_set = set(rev_dict.keys())
     ref_info_dict = exp_refs_by_baseline(repo.scm, rev_set, git_remote)

dvc/repo/experiments/show.py

@@ -117,7 +117,7 @@ def _build_rows(
     **kwargs,
 ) -> Iterator[tuple["CellT", ...]]:
     for baseline in baseline_states:
-        row: dict[str, "CellT"] = dict.fromkeys(all_headers, fill_value)
+        row: dict[str, CellT] = dict.fromkeys(all_headers, fill_value)
         row["Experiment"] = ""
         if baseline.name:
             row["rev"] = baseline.name
@@ -232,7 +232,7 @@ def _exp_range_rows(
         logger.debug("Returning tip commit for legacy checkpoint exp")
     exp = first(exp_range.revs)
     if exp:
-        row: dict[str, "CellT"] = dict.fromkeys(all_headers, fill_value)
+        row: dict[str, CellT] = dict.fromkeys(all_headers, fill_value)
         row["Experiment"] = exp.name or ""
         row["rev"] = exp.rev[:7] if Git.is_sha(exp.rev) else exp.rev
         row["typ"] = "branch_base" if is_base else "branch_commit"

dvc/repo/experiments/utils.py

@@ -199,7 +199,7 @@ def exp_commits(
     scm: "Git", ref_infos: Optional[Iterable[ExpRefInfo]] = None
 ) -> Iterable[str]:
     """Iterate over all experiment commits."""
-    shas: set["str"] = set()
+    shas: set[str] = set()
     refs = ref_infos if ref_infos else exp_refs(scm)
     for ref_info in refs:
         shas.update(scm.branch_revs(str(ref_info), ref_info.baseline_sha))

dvc/repo/fetch.py

@@ -2,6 +2,7 @@ from typing import TYPE_CHECKING
 
 from dvc.exceptions import DownloadError
 from dvc.log import logger
+from dvc.stage.cache import RunCacheNotSupported
 from dvc.ui import ui
 from dvc_data.index import DataIndex, FileStorage
 
@@ -136,6 +137,8 @@ def fetch(  # noqa: PLR0913
     try:
         if run_cache:
             self.stage_cache.pull(remote)
+    except RunCacheNotSupported as e:
+        logger.debug("failed to pull run cache: %s", e)
     except DownloadError as exc:
         failed_count += exc.amount
 
@@ -200,7 +203,7 @@ def fetch(  # noqa: PLR0913
 
 
 def _log_unversioned(data: list["DataIndex"]) -> tuple[list["DataIndex"], int]:
-    ret: list["DataIndex"] = []
+    ret: list[DataIndex] = []
     unversioned: list[str] = []
     for fs_index in data:
         remote = fs_index.storage_map[()].remote

dvc/repo/gc.py

@@ -50,7 +50,7 @@ def _used_obj_ids_not_in_remote(
 
 
 @locked
-def gc(  # noqa: PLR0913, C901
+def gc(  # noqa: C901, PLR0912, PLR0913
     self: "Repo",
     all_branches: bool = False,
     cloud: bool = False,
@@ -96,7 +96,7 @@ def gc(  # noqa: PLR0913, C901
         repos = []
     all_repos = [Repo(path) for path in repos]
 
-    odb_to_obj_ids: "ObjectContainer" = {}
+    odb_to_obj_ids: ObjectContainer = {}
     with ExitStack() as stack:
         for repo in all_repos:
             stack.enter_context(repo.lock)

dvc/repo/get.py

@@ -54,7 +54,7 @@ def get(
         from dvc.fs import download
         from dvc.fs.data import DataFileSystem
 
-        fs: Union[DataFileSystem, "DVCFileSystem"]
+        fs: Union[DataFileSystem, DVCFileSystem]
         if os.path.isabs(path):
             fs = DataFileSystem(index=repo.index.data["local"])
             fs_path = fs.from_os_path(path)

dvc/repo/graph.py

@@ -22,7 +22,7 @@ def check_acyclic(graph: "DiGraph") -> None:
     except nx.NetworkXNoCycle:
         return
 
-    stages: set["Stage"] = set()
+    stages: set[Stage] = set()
     for from_node, to_node, _ in edges:
         stages.add(from_node)
         stages.add(to_node)

dvc/repo/index.py

@@ -300,7 +300,7 @@ class Index:
         self._artifacts = artifacts or {}
         self._datasets: dict[str, list[dict[str, Any]]] = datasets or {}
         self._datasets_lock: dict[str, list[dict[str, Any]]] = datasets_lock or {}
-        self._collected_targets: dict[int, list["StageInfo"]] = {}
+        self._collected_targets: dict[int, list[StageInfo]] = {}
 
     @cached_property
     def rev(self) -> Optional[str]:
@@ -416,7 +416,7 @@ class Index:
 
     @cached_property
     def out_data_keys(self) -> dict[str, set["DataIndexKey"]]:
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         by_workspace["repo"] = set()
         by_workspace["local"] = set()
@@ -470,7 +470,7 @@ class Index:
 
     @cached_property
     def data_keys(self) -> dict[str, set["DataIndexKey"]]:
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         by_workspace["repo"] = set()
         by_workspace["local"] = set()
@@ -488,7 +488,7 @@ class Index:
     def metric_keys(self) -> dict[str, set["DataIndexKey"]]:
         from .metrics.show import _collect_top_level_metrics
 
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         by_workspace["repo"] = set()
 
@@ -509,7 +509,7 @@ class Index:
     def param_keys(self) -> dict[str, set["DataIndexKey"]]:
         from .params.show import _collect_top_level_params
 
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
         by_workspace["repo"] = set()
 
         param_paths = _collect_top_level_params(self.repo)
@@ -525,7 +525,7 @@ class Index:
 
     @cached_property
     def plot_keys(self) -> dict[str, set["DataIndexKey"]]:
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         by_workspace["repo"] = set()
 
@@ -548,7 +548,7 @@ class Index:
 
     @cached_property
     def data(self) -> "dict[str, DataIndex]":
-        prefix: "DataIndexKey"
+        prefix: DataIndexKey
         loaded = False
 
         index = self.repo.data_index
@@ -600,7 +600,7 @@ class Index:
         if not onerror:
 
             def onerror(_target, _exc):
-                raise
+                raise  # noqa: PLE0704
 
         targets = ensure_list(targets)
         if not targets:
@@ -627,7 +627,7 @@ class Index:
         jobs: Optional[int] = None,
         push: bool = False,
     ) -> "ObjectContainer":
-        used: "ObjectContainer" = defaultdict(set)
+        used: ObjectContainer = defaultdict(set)
         pairs = self.collect_targets(targets, recursive=recursive, with_deps=with_deps)
         for stage, filter_info in pairs:
             for odb, objs in stage.get_used_objs(
@@ -748,7 +748,7 @@ class IndexView:
 
     @cached_property
     def out_data_keys(self) -> dict[str, set["DataIndexKey"]]:
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         by_workspace["repo"] = set()
         by_workspace["local"] = set()
@@ -764,7 +764,7 @@ class IndexView:
 
     @cached_property
     def _data_prefixes(self) -> dict[str, "_DataPrefixes"]:
-        prefixes: dict[str, "_DataPrefixes"] = defaultdict(
+        prefixes: dict[str, _DataPrefixes] = defaultdict(
             lambda: _DataPrefixes(set(), set())
         )
         for out, filter_info in self._filtered_outs:
@@ -781,7 +781,7 @@ class IndexView:
 
     @cached_property
     def data_keys(self) -> dict[str, set["DataIndexKey"]]:
-        ret: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        ret: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         for out, filter_info in self._filtered_outs:
             if not out.use_cache:
@@ -811,7 +811,7 @@ class IndexView:
             except KeyError:
                 return False
 
-        data: dict[str, Union["DataIndex", "DataIndexView"]] = {}
+        data: dict[str, Union[DataIndex, DataIndexView]] = {}
         for workspace, data_index in self._index.data.items():
             if self.stages:
                 data[workspace] = view(data_index, partial(key_filter, workspace))

dvc/repo/ls.py

@@ -77,7 +77,7 @@ def _ls(
     recursive: Optional[bool] = None,
     dvc_only: bool = False,
 ):
-    fs: "DVCFileSystem" = repo.dvcfs
+    fs: DVCFileSystem = repo.dvcfs
     fs_path = fs.from_os_path(path)
 
     fs_path = fs.info(fs_path)["name"]

dvc/repo/metrics/show.py

@@ -224,7 +224,7 @@ def show(
                 on_error=on_error,
             )
             res[rev] = Result(data=result)
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             if on_error == "raise":
                 raise
 

dvc/repo/open_repo.py

@@ -146,7 +146,7 @@ def _cached_clone(url, rev):
 
 
 @wrap_with(threading.Lock())
-def _clone_default_branch(url, rev):  # noqa: PLR0912
+def _clone_default_branch(url, rev):
     """Get or create a clean clone of the url.
 
     The cloned is reactualized with git pull unless rev is a known sha.

dvc/repo/params/show.py

@@ -191,7 +191,7 @@ def show(
                 on_error=on_error,
             )
             res[rev] = Result(data=params)
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             if on_error == "raise":
                 raise
             logger.warning("failed to load params in revision %r, %s", rev, str(exc))

dvc/repo/push.py

@@ -1,10 +1,14 @@
 from contextlib import suppress
 
 from dvc.exceptions import InvalidArgumentError, UploadError
+from dvc.log import logger
+from dvc.stage.cache import RunCacheNotSupported
 from dvc.ui import ui
 
 from . import locked
 
+logger = logger.getChild(__name__)
+
 
 def _rebuild(idx, path, fs, cb):
     from dvc_data.index import DataIndex, DataIndexEntry, Meta
@@ -95,8 +99,11 @@ def push(  # noqa: PLR0913
                 "Multiple rev push is unsupported for cloud versioned remotes"
             )
 
-    used_run_cache = self.stage_cache.push(remote) if run_cache else []
-    transferred_count += len(used_run_cache)
+    try:
+        used_run_cache = self.stage_cache.push(remote) if run_cache else []
+        transferred_count += len(used_run_cache)
+    except RunCacheNotSupported as e:
+        logger.debug("failed to push run cache: %s", e)
 
     if isinstance(targets, str):
         targets = [targets]

dvc/repo/reproduce.py

@@ -29,7 +29,7 @@ def collect_stages(
     recursive: bool = False,
     glob: bool = False,
 ) -> list["Stage"]:
-    stages: list["Stage"] = []
+    stages: list[Stage] = []
     for target in targets:
         stages.extend(repo.stage.collect(target, recursive=recursive, glob=glob))
     return ldistinct(stages)
@@ -166,10 +166,10 @@ def _reproduce(
 ) -> list["Stage"]:
     assert on_error in ("fail", "keep-going", "ignore")
 
-    result: list["Stage"] = []
-    failed: list["Stage"] = []
-    to_skip: dict["Stage", "Stage"] = {}
-    ret: Optional["Stage"] = None
+    result: list[Stage] = []
+    failed: list[Stage] = []
+    to_skip: dict[Stage, Stage] = {}
+    ret: Optional[Stage] = None
 
     force_state = dict.fromkeys(stages, force)
 
@@ -228,7 +228,7 @@ def reproduce(
     if not kwargs.get("interactive", False):
         kwargs["interactive"] = self.config["core"].get("interactive", False)
 
-    stages: list["Stage"] = []
+    stages: list[Stage] = []
     if not all_pipelines:
         targets_list = ensure_list(targets or PROJECT_FILE)
         stages = collect_stages(self, targets_list, recursive=recursive, glob=glob)

dvc/repo/scm_context.py

@@ -21,7 +21,7 @@ class SCMContext:
     def __init__(self, scm: "Base", config: Optional[dict[str, Any]] = None) -> None:
         from funcy import get_in
 
-        self.scm: "Base" = scm
+        self.scm: Base = scm
         self.autostage: bool = get_in(
             config or {}, ["core", "autostage"], default=False
         )

dvc/repo/stage.py

@@ -108,7 +108,7 @@ def locked(f):
 
 class StageLoad:
     def __init__(self, repo: "Repo") -> None:
-        self.repo: "Repo" = repo
+        self.repo: Repo = repo
 
     @property
     def fs(self):

dvc/repo/update.py

@@ -34,7 +34,7 @@ def update(  # noqa: C901
         raise InvalidArgumentError("--remote can't be used without --to-remote")
 
     import_stages = set()
-    other_stage_infos: list["StageInfo"] = []
+    other_stage_infos: list[StageInfo] = []
 
     for stage_info in self.index.collect_targets(targets, recursive=recursive):
         if stage_info.stage.is_import:

dvc/repo/worktree.py

@@ -166,7 +166,7 @@ def update_worktree_stages(repo: "Repo", stage_infos: Iterable["StageInfo"]):
 
     view = IndexView(repo.index, stage_infos, outs_filter=outs_filter)
     local_index = view.data["repo"]
-    remote_indexes: dict[str, tuple["Remote", "DataIndex"]] = {}
+    remote_indexes: dict[str, tuple[Remote, DataIndex]] = {}
     for stage in view.stages:
         for out in stage.outs:
             _update_worktree_out(repo, out, local_index, remote_indexes)

dvc/stage/__init__.py

@@ -290,7 +290,7 @@ class Stage(params.StageParams):
         )
 
     def short_description(self) -> Optional["str"]:
-        desc: Optional["str"] = None
+        desc: Optional[str] = None
         if self.desc:
             with suppress(ValueError):
                 # try to use first non-empty line as a description
@@ -339,7 +339,7 @@ class Stage(params.StageParams):
         for out in self.outs:
             status = out.status()
             if status:
-                if allow_missing and status[str(out)] == "not in cache":
+                if allow_missing and status[str(out)] in ["not in cache", "deleted"]:
                     continue
                 logger.debug(
                     "Output '%s' of %s changed because it is '%s'.",
@@ -418,17 +418,28 @@ class Stage(params.StageParams):
 
     @rwlocked(read=["deps"], write=["outs"])
     def reproduce(self, interactive=False, **kwargs) -> Optional["Stage"]:
-        if not (
-            kwargs.get("force", False)
-            or self.changed(
-                kwargs.get("allow_missing", False), kwargs.pop("upstream", None)
-            )
-        ):
+        force = kwargs.get("force", False)
+        allow_missing = kwargs.get("allow_missing", False)
+        pull = kwargs.get("pull", False)
+        upstream = kwargs.pop("upstream", None)
+        if force:
+            pass
+        # Skip stages with missing data if otherwise unchanged
+        elif not self.changed(allow_missing, upstream):
             if not isinstance(self, PipelineStage) and self.is_data_source:
                 logger.info("'%s' didn't change, skipping", self.addressing)
             else:
                 logger.info("Stage '%s' didn't change, skipping", self.addressing)
             return None
+        # Pull stages with missing data if otherwise unchanged
+        elif not self.changed(True, upstream) and pull:
+            try:
+                logger.info("Pulling data for %s", self)
+                self.repo.pull(self.addressing, jobs=kwargs.get("jobs", None))
+                self.checkout()
+                return None
+            except CheckoutError:
+                logger.info("Unable to pull data for %s", self)
 
         msg = f"Going to reproduce {self}. Are you sure you want to continue?"
         if interactive and not prompt.confirm(msg):
@@ -597,16 +608,10 @@ class Stage(params.StageParams):
         if (self.cmd or self.is_import) and not self.frozen and not dry:
             self.remove_outs(ignore_remove=False, force=False)
 
-        if (
-            self.is_import and (not self.frozen or kwargs.get("pull"))
-        ) or self.is_partial_import:
+        if (self.is_import and not self.frozen) or self.is_partial_import:
             self._sync_import(dry, force, kwargs.get("jobs", None), no_download)
         elif not self.frozen and self.cmd:
             self._run_stage(dry, force, **kwargs)
-        elif kwargs.get("pull"):
-            logger.info("Pulling data for %s", self)
-            self.repo.pull(self.addressing, jobs=kwargs.get("jobs", None))
-            self.checkout()
         elif not dry:
             args = ("outputs", "frozen ") if self.frozen else ("data sources", "")
             logger.info("Verifying %s in %s%s", *args, self)
@@ -653,7 +658,7 @@ class Stage(params.StageParams):
     def checkout(
         self, allow_missing: bool = False, **kwargs
     ) -> dict[str, list["StrPath"]]:
-        stats: dict[str, list["StrPath"]] = defaultdict(list)
+        stats: dict[str, list[StrPath]] = defaultdict(list)
         if self.is_partial_import:
             return stats
 

dvc/stage/cache.py

@@ -229,7 +229,7 @@ class StageCache:
         if not dry:
             cached_stage.checkout()
 
-    def transfer(self, from_odb, to_odb):
+    def transfer(self, from_odb, to_odb, force=True):
         from dvc.fs import HTTPFileSystem, LocalFileSystem
         from dvc.fs.callbacks import TqdmCallback
 
@@ -258,10 +258,11 @@ class StageCache:
 
             dst = to_fs.join(to_odb.path, rel)
             key = to_fs.parent(dst)
+
             # check if any build cache already exists for this key
             # TODO: check if MaxKeys=1 or something like that applies
             # or otherwise this will take a lot of time!
-            if to_fs.exists(key) and first(to_fs.find(key)):
+            if not force and to_fs.exists(key) and first(to_fs.find(key)):
                 continue
 
             src_name = from_fs.name(src)

dvc/stage/loader.py

@@ -50,7 +50,7 @@ class StageLoader(Mapping):
         from dvc.output import Output, merge_file_meta_from_cloud
 
         assert isinstance(lock_data, dict)
-        items: Iterable[tuple[str, "Output"]] = chain(
+        items: Iterable[tuple[str, Output]] = chain(
             ((StageParams.PARAM_DEPS, dep) for dep in stage.deps),
             ((StageParams.PARAM_OUTS, out) for out in stage.outs),
         )

dvc/stage/run.py

@@ -2,6 +2,9 @@ import os
 import signal
 import subprocess
 import threading
+from functools import cache
+
+from packaging.version import InvalidVersion, Version
 
 from dvc.log import logger
 from dvc.utils import fix_env
@@ -12,25 +15,59 @@ from .exceptions import StageCmdFailedError
 logger = logger.getChild(__name__)
 
 
+@cache
+def _fish_supports_no_config(executable) -> bool:
+    """
+    Check if the fish shell supports the --no-config option.
+
+    Parameters:
+    executable (str): The path to the fish shell executable.
+
+    Returns:
+    bool: True if the fish version is greater than 3.3.0, False otherwise.
+    """
+    try:
+        output = subprocess.check_output(  # noqa: S603
+            [executable, "--version"],
+            text=True,
+        )
+        version = Version(output.split(" ")[-1].strip())
+        version_to_check = Version("3.3.0")
+        return version >= version_to_check
+    except (subprocess.CalledProcessError, IndexError, InvalidVersion):
+        logger.trace("could not check fish version, defaulting to False")
+        return False
+
+
+def _warn_if_fish(executable):
+    if (
+        executable is None
+        or os.path.basename(executable) != "fish"
+        or _fish_supports_no_config(executable)
+    ):
+        return
+    logger.warning(
+        "DVC detected that you are using a version of fish shell below 3.3.0 "
+        "Be aware that it might cause problems by overwriting "
+        "your current environment variables with values defined "
+        "in 'config.fish', which might affect your command. See "
+        "https://github.com/iterative/dvc/issues/1307. "
+    )
+
+
 def _make_cmd(executable, cmd):
     if executable is None:
         return cmd
-    opts = {"zsh": ["--no-rcs"], "bash": ["--noprofile", "--norc"]}
+    opts = {
+        "zsh": ["--no-rcs"],
+        "bash": ["--noprofile", "--norc"],
+        "fish": [],
+    }
     name = os.path.basename(executable).lower()
-    return [executable, *opts.get(name, []), "-c", cmd]
-
-
-def warn_if_fish(executable):
-    if executable is None or os.path.basename(os.path.realpath(executable)) != "fish":
-        return
-
-    logger.warning(
-        "DVC detected that you are using fish as your default "
-        "shell. Be aware that it might cause problems by overwriting "
-        "your current environment variables with values defined "
-        "in '.fishrc', which might affect your command. See "
-        "https://github.com/iterative/dvc/issues/1307. "
-    )
+    opt = opts.get(name, [])
+    if name == "fish" and _fish_supports_no_config(executable):
+        opt.append("--no-config")
+    return [executable, *opt, "-c", cmd]
 
 
 def _enforce_cmd_list(cmd):
@@ -106,7 +143,7 @@ def cmd_run(stage, dry=False, run_env=None):
     executable = get_executable()
 
     if not dry:
-        warn_if_fish(executable)
+        _warn_if_fish(executable)
 
     for cmd in commands:
         display_command(cmd)

dvc/testing/benchmarks/fixtures.py

@@ -104,6 +104,11 @@ def make_dvc_bin(
             venv.run("pip", "install", f"git+file://{dvc_git_repo}@{dvc_rev}#egg={egg}")
             if dvc_rev in ["2.18.1", "2.11.0", "2.6.3"]:
                 venv.run("pip", "install", "fsspec==2022.11.0")
+            try:
+                if version.Version(dvc_rev) < version.Version("3.50.3"):
+                    venv.run("pip", "install", "pygit2==1.14.1")
+            except version.InvalidVersion:
+                pass
             dvc_venvs[dvc_rev] = venv
         dvc_bin = venv.which("dvc")
     else:

dvc/testing/fixtures.py

@@ -228,10 +228,10 @@ def docker_services(tmp_path_factory, request):
         pytest.skip("disabled for Windows on CI")
 
     try:
-        subprocess.check_output(
+        subprocess.check_output(  # noqa: S602
             "docker ps",  # noqa: S607
             stderr=subprocess.STDOUT,
-            shell=True,  # noqa: S602
+            shell=True,
         )
     except subprocess.CalledProcessError as err:
         out = (err.output or b"").decode("utf-8")

dvc/testing/remote_tests.py

@@ -3,7 +3,6 @@ import shutil
 
 import pytest
 
-from dvc.stage.cache import RunCacheNotSupported
 from dvc.utils.fs import remove
 from dvc_data.hashfile.tree import Tree
 
@@ -97,8 +96,10 @@ class TestRemote:
         status_dir = dvc.cloud.status(dir_hashes)
         _check_status(status_dir, ok=dir_hashes)
 
-    @pytest.mark.xfail(raises=RunCacheNotSupported, strict=False)
     def test_stage_cache_push_pull(self, tmp_dir, dvc, remote):
+        if remote.scheme in ("http", "https"):
+            pytest.skip("HTTP remote does not support stage cache")
+
         tmp_dir.gen("foo", "foo")
         stage = dvc.stage.add(
             deps=["foo"], outs=["bar"], name="copy-foo-bar", cmd="cp foo bar"

dvc/ui/__init__.py

@@ -93,7 +93,7 @@ class Console:
             styled=styled,
         )
 
-    def write_json(  # noqa: PLR0913
+    def write_json(
         self,
         data: Any,
         indent: Optional[int] = None,

dvc/utils/studio.py

@@ -23,7 +23,7 @@ if TYPE_CHECKING:
 
 logger = logger.getChild(__name__)
 
-STUDIO_URL = "https://studio.iterative.ai"
+STUDIO_URL = "https://studio.dvc.ai"
 
 
 def post(

pyproject.toml

@@ -42,11 +42,11 @@ dependencies = [
     "dvc-http>=2.29.0",
     "dvc-objects",
     "dvc-render>=1.0.1,<2",
-    "dvc-studio-client>=0.20,<1",
+    "dvc-studio-client>=0.21,<1",
     "dvc-task>=0.3.0,<1",
     "flatten_dict<1,>=0.4.1",
-    "flufl.lock>=5,<8", # https://github.com/iterative/dvc/issues/9654
-    "fsspec",
+    "flufl.lock>=8.1.0,<9", # https://github.com/iterative/dvc/issues/9654
+    "fsspec>=2024.2.0",
     "funcy>=1.14",
     "grandalf<1,>=0.7",
     "gto>=1.6.0,<2",
@@ -65,7 +65,7 @@ dependencies = [
     "requests>=2.22",
     "rich>=12",
     "ruamel.yaml>=0.17.11",
-    "scmrepo>=3.3.2,<4",
+    "scmrepo>=3.3.4,<4",
     "shortuuid>=0.5",
     "shtab<2,>=1.3.4",
     "tabulate>=0.8.7",
@@ -83,7 +83,7 @@ gdrive = ["dvc-gdrive>=3,<4"]
 gs = ["dvc-gs>=3,<4"]
 hdfs = ["dvc-hdfs>=3,<4"]
 lint = [
-    "mypy==1.10.0",
+    "mypy==1.10.1",
     "pandas-stubs",
     "types-colorama",
     "types-psutil",
@@ -256,6 +256,7 @@ module = [
 
 [tool.codespell]
 ignore-words-list = "ba,datas,fo,uptodate,cachable,falsy"
+skip = "CODE_OF_CONDUCT.md"
 
 [tool.ruff]
 output-format = "full"

tests/func/artifacts/test_artifacts.py

@@ -281,3 +281,12 @@ def test_artifacts_download_subrepo(tmp_dir, scm, sub):
     assert Artifacts.get(subdir.fs_path, name, force=True) == result
     with subdir.chdir():
         assert Artifacts.get(".", name, force=True) == result
+
+
+def test_artifacts_download_studio(tmp_dir, dvc, mocker):
+    with dvc.config.edit("global") as conf:
+        conf["studio"]["token"] = "mytoken"
+
+    download_studio = mocker.patch("dvc.repo.artifacts.Artifacts._download_studio")
+    Artifacts.get("myart.pkl", "myart.pkl")
+    assert download_studio.call_args.kwargs["dvc_studio_config"]["token"] == "mytoken"

tests/func/repro/test_repro_allow_missing.py

@@ -55,3 +55,16 @@ def test_repro_allow_missing_upstream_stage_modified(
     ret = dvc.reproduce(pull=True, allow_missing=True)
     # create-foo is skipped ; copy-foo pulls modified dep
     assert len(ret) == 1
+
+
+def test_repro_allow_missing_cached(tmp_dir, dvc):
+    tmp_dir.gen("fixed", "fixed")
+    dvc.stage.add(name="create-foo", cmd="echo foo > foo", deps=["fixed"], outs=["foo"])
+    dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
+    dvc.reproduce()
+
+    remove("foo")
+
+    ret = dvc.reproduce(allow_missing=True)
+    # both stages are skipped
+    assert not ret

tests/func/repro/test_repro_pull.py

@@ -1,5 +1,7 @@
 import os
 
+import pytest
+
 from dvc.stage.cache import RunCacheNotSupported
 from dvc.utils.fs import remove
 
@@ -49,11 +51,14 @@ def test_repro_skip_pull_if_no_run_cache_is_passed(tmp_dir, dvc, mocker, local_r
     (foo,) = tmp_dir.dvc_gen("foo", "foo")
 
     dvc.push()
-    spy_pull = mocker.spy(dvc.stage_cache, "pull")
     dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
+
+    assert dvc.reproduce(pull=True)
+
     remove("foo")
     remove(foo.outs[0].cache_path)
-
+    remove("dvc.lock")
+    spy_pull = mocker.spy(dvc.stage_cache, "pull")
     assert dvc.reproduce(pull=True, run_cache=False)
     assert not spy_pull.called
 
@@ -68,3 +73,46 @@ def test_repro_skip_pull_if_single_item_is_passed(tmp_dir, dvc, mocker, local_re
     remove(foo.outs[0].cache_path)
 
     assert dvc.reproduce(pull=True, single_item=True)
+
+
+def test_repro_pulls_persisted_output(tmp_dir, dvc, mocker, local_remote):
+    tmp_dir.dvc_gen("foo", "foo")
+
+    dvc.push()
+
+    dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs_persist=["bar"])
+    dvc.reproduce()
+    remove("bar")
+
+    # stage is skipped
+    assert not dvc.reproduce(pull=True)
+
+
+@pytest.mark.parametrize("allow_missing", [True, False])
+def test_repro_pulls_allow_missing(tmp_dir, dvc, mocker, local_remote, allow_missing):
+    tmp_dir.dvc_gen("foo", "foo")
+
+    dvc.push()
+
+    dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
+    dvc.reproduce()
+    remove("foo")
+
+    # stage is skipped
+    assert not dvc.reproduce(pull=True, allow_missing=allow_missing)
+    # data only pulled if allow_missing is false
+    assert (tmp_dir / "foo").exists() != allow_missing
+
+
+def test_repro_pull_fails(tmp_dir, dvc, mocker, local_remote):
+    tmp_dir.dvc_gen("foo", "foo")
+
+    dvc.stage.add(
+        name="concat-foo", cmd="cat foo foo > bar", deps=["foo"], outs=["bar"]
+    )
+    stages = dvc.reproduce()
+    remove("bar")
+    remove(stages[0].outs[0].cache_path)
+
+    # stage is run
+    assert dvc.reproduce(pull=True)

tests/func/test_download.py

@@ -0,0 +1,73 @@
+import os
+from unittest.mock import ANY
+
+import pytest
+
+from dvc.fs import download
+
+lfs_prefetch_params = [
+    pytest.param("abc", "abc", id="plain"),
+    pytest.param(
+        "*",
+        "[*]",
+        marks=pytest.mark.skipif(
+            os.name == "nt",
+            reason="forbidden character `*` on Windows filesystem",
+        ),
+        id="escape-*",
+    ),
+    pytest.param(
+        "**",
+        "[*][*]",
+        marks=pytest.mark.skipif(
+            os.name == "nt", reason="forbidden character `*` on Windows filesystem"
+        ),
+        id="escape-**",
+    ),
+    pytest.param(
+        "?",
+        "[?]",
+        marks=pytest.mark.skipif(
+            os.name == "nt", reason="forbidden character `?` on Windows filesystem"
+        ),
+        id="escape-?",
+    ),
+    pytest.param("[abc]", "[[]abc]", id="escape-[seq]"),
+    pytest.param("[!abc]", "[[]!abc]", id="escape-[!seq]"),
+]
+
+
+@pytest.mark.parametrize("dirname, include_name", lfs_prefetch_params)
+def test_lfs_prefetch_directory(tmp_dir, dvc, scm, mocker, dirname, include_name):
+    mock_fetch = mocker.patch("scmrepo.git.lfs.fetch")
+    tmp_dir.scm_gen(
+        {
+            ".gitattributes": "data/**/* filter=lfs diff=lfs merge=lfs -text",
+            f"data/{dirname}/test.txt": "test data",
+        },
+        commit="init lfs",
+    )
+    rev = scm.get_rev()
+    with dvc.switch(rev):
+        download(dvc.dvcfs, f"data/{dirname}", "data")
+        mock_fetch.assert_called_once_with(
+            scm, [rev], include=[f"/data/{include_name}/**"], progress=ANY
+        )
+
+
+@pytest.mark.parametrize("basename, include_name", lfs_prefetch_params)
+def test_lfs_prefetch_file(tmp_dir, dvc, scm, mocker, basename, include_name):
+    mock_fetch = mocker.patch("scmrepo.git.lfs.fetch")
+    tmp_dir.scm_gen(
+        {
+            ".gitattributes": "data/**/* filter=lfs diff=lfs merge=lfs -text",
+            f"data/{basename}.txt": "test data",
+        },
+        commit="init lfs",
+    )
+    rev = scm.get_rev()
+    with dvc.switch(rev):
+        download(dvc.dvcfs, f"data/{basename}.txt", "data")
+        mock_fetch.assert_called_once_with(
+            scm, [rev], include=[f"/data/{include_name}.txt"], progress=ANY
+        )

tests/func/test_import.py

@@ -305,11 +305,12 @@ def test_cache_type_is_properly_overridden(tmp_dir, scm, dvc, erepo_dir):
     assert scm.is_ignored("foo_imported")
 
 
-def test_pull_imported_directory_stage(tmp_dir, dvc, erepo_dir):
+@pytest.mark.parametrize("dirpath", ["dir", "dir/"])
+def test_pull_imported_directory_stage(tmp_dir, dvc, erepo_dir, dirpath):
     with erepo_dir.chdir():
         erepo_dir.dvc_gen({"dir": {"foo": "foo content"}}, commit="create dir")
 
-    dvc.imp(os.fspath(erepo_dir), "dir", "dir_imported")
+    stage = dvc.imp(os.fspath(erepo_dir), dirpath, "dir_imported")
 
     remove("dir_imported")
     dvc.cache.local.clear()
@@ -317,6 +318,7 @@ def test_pull_imported_directory_stage(tmp_dir, dvc, erepo_dir):
     dvc.pull(["dir_imported.dvc"])
 
     assert (tmp_dir / "dir_imported").read_text() == {"foo": "foo content"}
+    assert stage.deps[0].fs_path == "dir"
 
 
 def test_pull_wildcard_imported_directory_stage(tmp_dir, dvc, erepo_dir):

tests/func/test_run_cache.py

@@ -1,3 +1,4 @@
+import logging
 import os
 
 import pytest
@@ -231,13 +232,15 @@ def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote):
     assert (tmp_dir / LOCK_FILE).exists()
 
 
-def test_push_pull_unsupported(tmp_dir, dvc, mocker, run_copy):
+def test_push_pull_unsupported(tmp_dir, dvc, mocker, run_copy, local_remote, caplog):
     tmp_dir.gen("foo", "foo")
     run_copy("foo", "bar", name="copy-foo-bar")
     mocker.patch.object(
         dvc.cloud, "get_remote_odb", side_effect=RunCacheNotSupported("foo")
     )
-    with pytest.raises(RunCacheNotSupported):
+    with caplog.at_level(logging.DEBUG):
         dvc.push(run_cache=True)
-    with pytest.raises(RunCacheNotSupported):
+        assert "failed to push run cache" in caplog.text
+    with caplog.at_level(logging.DEBUG):
         dvc.pull(run_cache=True)
+        assert "failed to pull run cache" in caplog.text