Class DVCFileSystem

dvc/fs/dvc.py:667–753

class DVCFileSystem(FileSystem):
    protocol = "local"
    PARAM_CHECKSUM = "md5"

    def _prepare_credentials(self, **config) -> dict[str, Any]:
        return config

    @functools.cached_property
    def fs(self) -> "_DVCFileSystem":
        return _DVCFileSystem(**self.fs_args)

    @property
    def immutable(self):
        from dvc.scm import NoSCM

        if isinstance(self.fs.repo.scm, NoSCM):
            return False

        return self.fs._repo_kwargs.get("rev") == self.fs.repo.get_rev()

    def getcwd(self):
        return self.fs.getcwd()

    def _get(
        self,
        from_info: Union[AnyFSPath, list[AnyFSPath]],
        to_info: Union[AnyFSPath, list[AnyFSPath]],
        callback: "Callback" = DEFAULT_CALLBACK,
        recursive: bool = False,
        batch_size: Optional[int] = None,
        **kwargs,
    ) -> list[tuple[str, str, Optional[dict]]]:
        # FileSystem.get is non-recursive by default if arguments are lists
        # otherwise, it's recursive.
        recursive = not (isinstance(from_info, list) and isinstance(to_info, list))
        return self.fs._get(
            from_info,
            to_info,
            callback=callback,
            recursive=recursive,
            batch_size=batch_size,
            **kwargs,
        )

    def get(
        self,
        from_info: Union[AnyFSPath, list[AnyFSPath]],
        to_info: Union[AnyFSPath, list[AnyFSPath]],
        callback: "Callback" = DEFAULT_CALLBACK,
        recursive: bool = False,
        batch_size: Optional[int] = None,
        **kwargs,
    ) -> None:
        self._get(
            from_info,
            to_info,
            callback=callback,
            batch_size=batch_size,