diff --git a/pyproject.toml b/pyproject.toml index 43be8e1..cfb6670 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,7 +30,7 @@ dependencies = [ "asyncssh>=2.13.1,<3", "funcy>=1.14", "shortuuid>=0.5.0", - "dvc-objects>=1.0.1,<3", + "dvc-objects>=3,<4", "dvc-http>=2.29.0", ] diff --git a/src/scmrepo/fs.py b/src/scmrepo/fs.py index 7de65cd..850d113 100644 --- a/src/scmrepo/fs.py +++ b/src/scmrepo/fs.py @@ -14,49 +14,79 @@ from scmrepo.git.objects import GitTrie -class Path: - def __init__(self, sep, getcwd=None, realpath=None): - def _getcwd(): - return "" +def bytesio_len(obj: "BytesIO") -> Optional[int]: + try: + offset = obj.tell() + length = obj.seek(0, os.SEEK_END) + obj.seek(offset) + except (AttributeError, OSError): + return None + return length - self.getcwd = getcwd or _getcwd - self.realpath = realpath or self.abspath - assert sep == posixpath.sep - self.flavour = posixpath +class GitFileSystem(AbstractFileSystem): + # pylint: disable=abstract-method + cachable = False + root_marker = "/" - def chdir(self, path): - def _getcwd(): - return path + def __init__( + self, + path: str = None, + rev: str = None, + scm: "Git" = None, + trie: "GitTrie" = None, + rev_resolver: Callable[["Git", str], str] = None, + **kwargs, + ): + from scmrepo.git import Git + from scmrepo.git.objects import GitTrie + + super().__init__(**kwargs) + if not trie: + scm = scm or Git(path) + resolver = rev_resolver or Git.resolve_rev + resolved = resolver(scm, rev or "HEAD") + tree_obj = scm.pygit2.get_tree_obj(rev=resolved) + trie = GitTrie(tree_obj, resolved) + + self.trie = trie + self.rev = self.trie.rev - self.getcwd = _getcwd + def getcwd(self): + return self.root_marker - def join(self, *parts): - return self.flavour.join(*parts) + def chdir(self, path): + raise NotImplementedError + + @classmethod + def join(cls, *parts): + return posixpath.join(*parts) - def split(self, path): - return self.flavour.split(path) + @classmethod + def split(cls, path): + return posixpath.split(path) def normpath(self, path): - return self.flavour.normpath(path) + return posixpath.normpath(path) - def isabs(self, path): - return self.flavour.isabs(path) + @classmethod + def isabs(cls, path): + return posixpath.isabs(path) def abspath(self, path): if not self.isabs(path): path = self.join(self.getcwd(), path) return self.normpath(path) - def commonprefix(self, path): - return self.flavour.commonprefix(path) - - def parts(self, path): - drive, path = self.flavour.splitdrive(path.rstrip(self.flavour.sep)) + @classmethod + def commonprefix(cls, path): + return posixpath.commonprefix(path) + @classmethod + def parts(cls, path): ret = [] while True: - path, part = self.flavour.split(path) + path, part = cls.split(path) if part: ret.append(part) @@ -69,113 +99,77 @@ def parts(self, path): ret.reverse() - if drive: - ret = [drive] + ret - return tuple(ret) - def parent(self, path): - return self.flavour.dirname(path) + @classmethod + def parent(cls, path): + return posixpath.dirname(path) - def dirname(self, path): - return self.parent(path) + @classmethod + def dirname(cls, path): + return cls.parent(path) - def parents(self, path): - parts = self.parts(path) + @classmethod + def parents(cls, path): + parts = cls.parts(path) return tuple( - self.join(*parts[:length]) for length in range(len(parts) - 1, 0, -1) + cls.join(*parts[:length]) for length in range(len(parts) - 1, 0, -1) ) - def name(self, path): - return self.parts(path)[-1] + @classmethod + def name(cls, path): + return cls.parts(path)[-1] - def suffix(self, path): - name = self.name(path) + @classmethod + def suffix(cls, path): + name = cls.name(path) _, dot, suffix = name.partition(".") return dot + suffix - def with_name(self, path, name): - parts = list(self.parts(path)) + @classmethod + def with_name(cls, path, name): + parts = list(cls.parts(path)) parts[-1] = name - return self.join(*parts) + return cls.join(*parts) - def with_suffix(self, path, suffix): - parts = list(self.parts(path)) + @classmethod + def with_suffix(cls, path, suffix): + parts = list(cls.parts(path)) real_path, _, _ = parts[-1].partition(".") parts[-1] = real_path + suffix - return self.join(*parts) + return cls.join(*parts) - def isin(self, left, right): - left_parts = self.parts(left) - right_parts = self.parts(right) + @classmethod + def isin(cls, left, right): + left_parts = cls.parts(left) + right_parts = cls.parts(right) left_len = len(left_parts) right_len = len(right_parts) return left_len > right_len and left_parts[:right_len] == right_parts - def isin_or_eq(self, left, right): - return left == right or self.isin(left, right) + @classmethod + def isin_or_eq(cls, left, right): + return left == right or cls.isin(left, right) - def overlaps(self, left, right): + @classmethod + def overlaps(cls, left, right): # pylint: disable=arguments-out-of-order - return self.isin_or_eq(left, right) or self.isin(right, left) + return cls.isin_or_eq(left, right) or cls.isin(right, left) def relpath(self, path, start=None): if start is None: start = "." - return self.flavour.relpath(self.abspath(path), start=self.abspath(start)) + return self.relpath(self.abspath(path), start=self.abspath(start)) def relparts(self, path, start=None): return self.parts(self.relpath(path, start=start)) - def as_posix(self, path): - return path.replace(self.flavour.sep, posixpath.sep) - - -def bytesio_len(obj: "BytesIO") -> Optional[int]: - try: - offset = obj.tell() - length = obj.seek(0, os.SEEK_END) - obj.seek(offset) - except (AttributeError, OSError): - return None - return length - - -class GitFileSystem(AbstractFileSystem): - # pylint: disable=abstract-method - cachable = False - root_marker = "/" - - def __init__( - self, - path: str = None, - rev: str = None, - scm: "Git" = None, - trie: "GitTrie" = None, - rev_resolver: Callable[["Git", str], str] = None, - **kwargs, - ): - from scmrepo.git import Git - from scmrepo.git.objects import GitTrie - - super().__init__(**kwargs) - if not trie: - scm = scm or Git(path) - resolver = rev_resolver or Git.resolve_rev - resolved = resolver(scm, rev or "HEAD") - tree_obj = scm.pygit2.get_tree_obj(rev=resolved) - trie = GitTrie(tree_obj, resolved) - - self.trie = trie - self.rev = self.trie.rev - - def _getcwd(): - return self.root_marker - - self.path = Path(self.sep, getcwd=_getcwd) + @classmethod + def as_posix(cls, path): + return path def _get_key(self, path: str) -> Tuple[str, ...]: - path = self.path.abspath(path) + path = self.abspath(path) if path == self.root_marker: return () relparts = path.split(self.sep)