mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-10 15:28:34 +01:00
dependabot[bot]andPeter Rowlands
6c14bf39dc
* build(deps): bump pylint from 2.8.3 to 2.9.6 Bumps [pylint](https://github.com/PyCQA/pylint) from 2.8.3 to 2.9.6. - [Release notes](https://github.com/PyCQA/pylint/releases) - [Changelog](https://github.com/PyCQA/pylint/blob/main/ChangeLog) - [Commits](https://github.com/PyCQA/pylint/compare/v2.8.3...v2.9.6) --- updated-dependencies: - dependency-name: pylint dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <[email protected]> * pylint: fix 2.9.6 issues Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Peter Rowlands <[email protected]>
129 lines
3.8 KiB
Python
129 lines
3.8 KiB
Python
import errno
|
|
import os
|
|
import stat
|
|
|
|
from dvc.utils import is_exec, relpath
|
|
|
|
from .base import BaseFileSystem
|
|
|
|
|
|
class GitFileSystem(BaseFileSystem): # pylint:disable=abstract-method
|
|
"""Proxies the repo file access methods to Git objects"""
|
|
|
|
sep = os.sep
|
|
|
|
scheme = "local"
|
|
|
|
def __init__(self, root_dir, trie):
|
|
super().__init__()
|
|
self._root = root_dir
|
|
self.trie = trie
|
|
|
|
@property
|
|
def rev(self):
|
|
return self.trie.rev
|
|
|
|
def _get_key(self, path):
|
|
if isinstance(path, str):
|
|
if not os.path.isabs(path):
|
|
relparts = path.split(os.sep)
|
|
else:
|
|
relparts = relpath(path, self._root).split(os.sep)
|
|
else:
|
|
relparts = path.relative_to(self._root).parts
|
|
if relparts == ["."]:
|
|
return ()
|
|
return tuple(relparts)
|
|
|
|
def open(
|
|
self, path, mode="r", encoding=None, **kwargs
|
|
): # pylint: disable=arguments-renamed
|
|
# NOTE: this is incorrect, we need to use locale to determine default
|
|
# encoding.
|
|
encoding = encoding or "utf-8"
|
|
|
|
key = self._get_key(path)
|
|
try:
|
|
return self.trie.open(key, mode=mode, encoding=encoding)
|
|
except KeyError as exc:
|
|
msg = os.strerror(errno.ENOENT) + f"in branch '{self.rev}'"
|
|
raise FileNotFoundError(errno.ENOENT, msg, path) from exc
|
|
except IsADirectoryError as exc:
|
|
raise IsADirectoryError(
|
|
errno.EISDIR, os.strerror(errno.EISDIR), path
|
|
) from exc
|
|
|
|
def exists(self, path_info) -> bool:
|
|
key = self._get_key(path_info)
|
|
return self.trie.exists(key)
|
|
|
|
def isdir(self, path_info) -> bool:
|
|
key = self._get_key(path_info)
|
|
return self.trie.isdir(key)
|
|
|
|
def isfile(self, path_info) -> bool:
|
|
key = self._get_key(path_info)
|
|
return self.trie.isfile(key)
|
|
|
|
def walk(self, top, topdown=True, onerror=None, **kwargs):
|
|
"""Directory tree generator.
|
|
|
|
See `os.walk` for the docs. Differences:
|
|
- no support for symlinks
|
|
"""
|
|
if not self.isdir(top):
|
|
if onerror:
|
|
if self.exists(top):
|
|
exc = NotADirectoryError(
|
|
errno.ENOTDIR, os.strerror(errno.ENOTDIR), top
|
|
)
|
|
else:
|
|
exc = FileNotFoundError(
|
|
errno.ENOENT, os.strerror(errno.ENOENT), top
|
|
)
|
|
onerror(exc)
|
|
return []
|
|
|
|
key = self._get_key(top)
|
|
for prefix, dirs, files in self.trie.walk(key, topdown=topdown):
|
|
if prefix:
|
|
root = os.path.join(self._root, os.sep.join(prefix))
|
|
else:
|
|
root = self._root
|
|
yield root, dirs, files
|
|
|
|
def isexec(self, path_info):
|
|
if not self.exists(path_info):
|
|
return False
|
|
|
|
mode = self.stat(path_info).st_mode
|
|
return is_exec(mode)
|
|
|
|
def info(self, path_info):
|
|
key = self._get_key(path_info)
|
|
try:
|
|
st = self.trie.stat(key)
|
|
except KeyError:
|
|
raise FileNotFoundError(
|
|
errno.ENOENT, os.strerror(errno.ENOENT), path_info
|
|
)
|
|
return {
|
|
"size": st.st_size,
|
|
"type": "directory" if stat.S_ISDIR(st.st_mode) else "file",
|
|
}
|
|
|
|
def stat(self, path):
|
|
key = self._get_key(path)
|
|
try:
|
|
return self.trie.stat(key)
|
|
except KeyError:
|
|
raise FileNotFoundError(
|
|
errno.ENOENT, os.strerror(errno.ENOENT), path
|
|
)
|
|
|
|
def walk_files(self, path_info, **kwargs):
|
|
for root, _, files in self.walk(path_info, **kwargs):
|
|
for file in files:
|
|
# NOTE: os.path.join is ~5.5 times slower
|
|
yield f"{root}{os.sep}{file}"
|