remove dvc.scm and dvc.fs.git module; rename dvc.fs.scm to dvc.fs.git

This commit is contained in:
Saugat Pachhai (सौगात)
2021-11-25 19:09:58 +02:00
committed by Ruslan Kuprieiev
parent 57c6b1036f
commit ea52ab2260
42 changed files with 121 additions and 3568 deletions
+2 -1
View File
@@ -82,9 +82,10 @@ def send(path):
def _scm_in_use():
from scmrepo.noscm import NoSCM
from dvc.exceptions import NotDvcRepoError
from dvc.repo import Repo
from dvc.scm.noscm import NoSCM
from .scm import SCM, SCMError
+1 -1
View File
@@ -142,7 +142,7 @@ def _collect_rows(
fill_value=FILL_VALUE,
iso=False,
):
from dvc.scm.git import Git
from scmrepo.git import Git
if sort_by:
sort_path, sort_name, sort_type = _sort_column(
+4 -3
View File
@@ -27,8 +27,9 @@ logger = logging.getLogger(__name__)
def external_repo(
url, rev=None, for_write=False, cache_dir=None, cache_types=None, **kwargs
):
from scmrepo.git import Git
from dvc.config import NoRemoteError
from dvc.scm.git import Git
logger.debug("Creating external repo %s@%s", url, rev)
path = _cached_clone(url, rev, for_write=for_write)
@@ -175,7 +176,7 @@ def _clone_default_branch(url, rev, for_write=False):
The cloned is reactualized with git pull unless rev is a known sha.
"""
from dvc.scm.git import Git
from scmrepo.git import Git
clone_path, shallow = CLONES.get(url, (None, False))
@@ -240,7 +241,7 @@ def _unshallow(git):
def _git_checkout(repo_path, rev):
from dvc.scm.git import Git
from scmrepo.git import Git
logger.debug("erepo: git checkout %s@%s", repo_path, rev)
git = Git(repo_path)
+46 -136
View File
@@ -1,38 +1,22 @@
import errno
import os
from typing import (
TYPE_CHECKING,
Any,
BinaryIO,
Callable,
Dict,
Optional,
Tuple,
)
import threading
from typing import TYPE_CHECKING, Any, Callable
from fsspec.spec import AbstractFileSystem
from funcy import cached_property, wrap_prop
from .fsspec_wrapper import AnyFSPath, FSSpecWrapper
if TYPE_CHECKING:
from io import BytesIO
from dvc.scm.git import Git
from dvc.scm.git.objects import GitTrie
from scmrepo.fs import GitFileSystem as FsspecGitFileSystem
from scmrepo.git import Git
from scmrepo.git.objects import GitTrie
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(FSSpecWrapper): # pylint:disable=abstract-method
"""Proxies the repo file access methods to Git objects"""
class GitFileSystem(AbstractFileSystem):
# pylint: disable=abstract-method
sep = os.sep
cachable = False
scheme = "local"
def __init__(
self,
@@ -40,123 +24,49 @@ class GitFileSystem(AbstractFileSystem):
rev: str = None,
scm: "Git" = None,
trie: "GitTrie" = None,
rev_resolver: Callable[["Git", str], str] = None,
**kwargs,
):
from dvc.scm.git import Git
from dvc.scm.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)
path = scm.root_dir
else:
assert path
self.trie = trie
self.root_dir = path
self.rev = self.trie.rev
def _get_key(self, path: str) -> Tuple[str, ...]:
from dvc.scm.utils import relpath
if os.path.isabs(path):
path = relpath(path, self.root_dir)
relparts = path.split(os.sep)
if relparts == ["."]:
return ()
return tuple(relparts)
def _open(
self,
path: str,
mode: str = "rb",
block_size: int = None,
autocommit: bool = True,
cache_options: Dict = None,
**kwargs: Any,
) -> BinaryIO:
key = self._get_key(path)
try:
obj = self.trie.open(key, mode=mode)
obj.size = bytesio_len(obj)
return obj
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
) -> None:
from dvc.scm import resolve_rev
def info(self, path: str, **kwargs: Any) -> Dict[str, Any]:
key = self._get_key(path)
try:
return {
**self.trie.info(key),
"name": os.path.join(self.root_dir, self.sep.join(key)),
super().__init__()
self.fs_args.update(
{
"path": path,
"rev": rev,
"scm": scm,
"trie": trie,
"rev_resolver": resolve_rev,
**kwargs,
}
except KeyError:
raise FileNotFoundError(
errno.ENOENT, os.strerror(errno.ENOENT), path
)
)
def exists(self, path: str, **kwargs: Any) -> bool:
key = self._get_key(path)
return self.trie.exists(key)
@wrap_prop(threading.Lock())
@cached_property
def fs(self) -> "FsspecGitFileSystem":
from scmrepo.fs import GitFileSystem as FsspecGitFileSystem
def checksum(self, path: str) -> str:
return self.info(path)["sha"]
return FsspecGitFileSystem(**self.fs_args)
def walk( # pylint: disable=arguments-differ
@property
def rev(self) -> str:
return self.fs.rev
def isexec(self, path: AnyFSPath) -> bool:
from dvc.utils import is_exec
try:
return is_exec(self.info(path)["mode"])
except FileNotFoundError:
return False
def isfile(self, path: AnyFSPath) -> bool:
return self.fs.isfile(path)
def walk(
self,
top: str,
top: AnyFSPath,
topdown: bool = True,
onerror: Callable[[OSError], None] = None,
maxdepth: int = None,
detail: bool = False,
**kwargs: Any,
):
"""Directory tree generator.
See `os.walk` for the docs. Differences:
- no support for symlinks
"""
assert maxdepth is None # not supported yet.
if not self.isdir(top):
if onerror:
if self.exists(top):
exc: OSError = 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):
root = self.root_dir
if prefix:
root = os.path.join(root, os.sep.join(prefix))
if detail:
yield (
root,
{d: self.info(os.path.join(root, d)) for d in dirs},
{f: self.info(os.path.join(root, f)) for f in files},
)
else:
yield root, dirs, files
def ls(self, path, detail=True, **kwargs):
for _, dirs, files in self.walk(path, detail=detail, **kwargs):
merge = files.update if detail else files.extend
merge(dirs)
return files
return self.fs.walk(top, topdown=topdown, onerror=onerror, **kwargs)
-72
View File
@@ -1,72 +0,0 @@
import os
import threading
from typing import TYPE_CHECKING, Any, Callable
from funcy import cached_property, wrap_prop
from .fsspec_wrapper import AnyFSPath, FSSpecWrapper
if TYPE_CHECKING:
from dvc.fs.git import GitFileSystem as FsspecGitFileSystem
from dvc.scm.git import Git
from dvc.scm.git.objects import GitTrie
class GitFileSystem(FSSpecWrapper): # pylint:disable=abstract-method
"""Proxies the repo file access methods to Git objects"""
sep = os.sep
scheme = "local"
def __init__(
self,
path: str = None,
rev: str = None,
scm: "Git" = None,
trie: "GitTrie" = None,
**kwargs: Any,
) -> None:
from dvc.scm import resolve_rev
super().__init__()
self.fs_args.update(
{
"path": path,
"rev": rev,
"scm": scm,
"trie": trie,
"rev_resolver": resolve_rev,
**kwargs,
}
)
@wrap_prop(threading.Lock())
@cached_property
def fs(self) -> "FsspecGitFileSystem":
from dvc.fs.git import GitFileSystem as FsspecGitFileSystem
return FsspecGitFileSystem(**self.fs_args)
@property
def rev(self) -> str:
return self.fs.rev
def isexec(self, path: AnyFSPath) -> bool:
from dvc.utils import is_exec
try:
return is_exec(self.info(path)["mode"])
except FileNotFoundError:
return False
def isfile(self, path: AnyFSPath) -> bool:
return self.fs.isfile(path)
def walk(
self,
top: AnyFSPath,
topdown: bool = True,
onerror: Callable[[OSError], None] = None,
**kwargs: Any,
):
return self.fs.walk(top, topdown=topdown, onerror=onerror, **kwargs)
+2 -2
View File
@@ -90,7 +90,7 @@ class Repo:
):
assert bool(scm) == bool(rev)
from dvc.fs.scm import GitFileSystem
from dvc.fs.git import GitFileSystem
from dvc.scm import SCM, Base, Git, SCMError
from dvc.utils.fs import makedirs
@@ -161,8 +161,8 @@ class Repo:
):
from dvc.config import Config
from dvc.data_cloud import DataCloud
from dvc.fs.git import GitFileSystem
from dvc.fs.local import LocalFileSystem
from dvc.fs.scm import GitFileSystem
from dvc.lock import LockNoop, make_lock
from dvc.objects.db import ODBManager
from dvc.repo.live import Live
+1 -1
View File
@@ -60,7 +60,7 @@ def brancher( # noqa: E302
try:
if revs:
from dvc.fs.scm import GitFileSystem
from dvc.fs.git import GitFileSystem
from dvc.scm import resolve_rev
rev_resolver = partial(resolve_rev, scm)
+1 -1
View File
@@ -119,7 +119,7 @@ class Experiments:
@cached_property
def stash(self):
from dvc.scm.git import Stash
from scmrepo.git import Stash
return Stash(self.scm, EXPS_STASH)
+3 -2
View File
@@ -19,9 +19,10 @@ logger = logging.getLogger(__name__)
@locked
@scm_context
def apply(repo, rev, force=True, **kwargs):
from scmrepo.exceptions import MergeConflictError
from dvc.repo.checkout import checkout as dvc_checkout
from dvc.scm import RevError, SCMError, resolve_rev
from dvc.scm.exceptions import MergeConflictError
exps = repo.experiments
@@ -45,7 +46,7 @@ def apply(repo, rev, force=True, **kwargs):
else:
workspace = None
from dvc.scm.exceptions import SCMError as _SCMError
from scmrepo.exceptions import SCMError as _SCMError
try:
repo.scm.merge(exp_rev, commit=False)
+3 -2
View File
@@ -38,7 +38,7 @@ from dvc.utils.fs import remove
if TYPE_CHECKING:
from multiprocessing import Queue
from dvc.scm.git import Git
from scmrepo.git import Git
logger = logging.getLogger(__name__)
@@ -224,8 +224,9 @@ class BaseExecutor(ABC):
@classmethod
def _validate_remotes(cls, dvc: "Repo", git_remote: Optional[str]):
from scmrepo.exceptions import InvalidRemote
from dvc.scm import InvalidRemoteSCMRepo
from dvc.scm.exceptions import InvalidRemote
if git_remote == dvc.root_dir:
logger.warning(
+2 -1
View File
@@ -17,8 +17,9 @@ from dvc.scm import SCM
from .base import BaseExecutor
if TYPE_CHECKING:
from scmrepo.git import Git
from dvc.repo import Repo
from dvc.scm.git import Git
logger = logging.getLogger(__name__)
+2 -1
View File
@@ -17,9 +17,10 @@ from dvc.repo.experiments.base import (
from .base import BaseExecutor
if TYPE_CHECKING:
from scmrepo.git import Git
from dvc.machine import MachineManager
from dvc.repo import Repo
from dvc.scm.git import Git
logger = logging.getLogger(__name__)
+2 -1
View File
@@ -17,8 +17,9 @@ logger = logging.getLogger(__name__)
@locked
@scm_context
def ls(repo, *args, rev=None, git_remote=None, all_=False, **kwargs):
from scmrepo.git import Git
from dvc.scm import RevError, resolve_rev
from dvc.scm.git import Git
if rev:
try:
+5 -3
View File
@@ -1,7 +1,8 @@
from typing import Callable, Generator, Iterable, Optional, Set
from scmrepo.git import Git
from dvc.exceptions import InvalidArgumentError
from dvc.scm.git import Git
from .base import (
EXEC_BASELINE,
@@ -52,8 +53,9 @@ def exp_refs_by_baseline(
def iter_remote_refs(
scm: "Git", url: str, base: Optional[str] = None, **kwargs
):
from scmrepo.exceptions import AuthError, InvalidRemote
from dvc.scm import GitAuthError, InvalidRemoteSCMRepo
from dvc.scm.exceptions import AuthError, InvalidRemote
try:
yield from scm.iter_remote_refs(url, base=base, **kwargs)
@@ -72,7 +74,7 @@ def push_refspec(
on_diverged: Optional[Callable[[str, str], bool]] = None,
**kwargs,
):
from dvc.scm.exceptions import AuthError
from scmrepo.exceptions import AuthError
from ...scm import GitAuthError
+5 -3
View File
@@ -3,8 +3,9 @@ from typing import TYPE_CHECKING
from dvc.exceptions import DvcException
if TYPE_CHECKING:
from scmrepo.git import Git
from dvc.repo import Repo
from dvc.scm.git import Git
def pre_commit_install(scm: "Git") -> None:
@@ -43,7 +44,8 @@ def pre_commit_install(scm: "Git") -> None:
def install_hooks(scm: "Git") -> None:
from dvc.scm.exceptions import GitHookAlreadyExists
from scmrepo.exceptions import GitHookAlreadyExists
from dvc.utils import format_link
hooks = ["post-checkout", "pre-commit", "pre-push"]
@@ -64,7 +66,7 @@ def install(self: "Repo", use_pre_commit_tool: bool = False) -> None:
If use_pre_commit_tool is set and pre-commit is installed it will be used
to install the hooks.
"""
from dvc.scm.git import Git
from scmrepo.git import Git
scm = self.scm
if not isinstance(scm, Git):
+7 -4
View File
@@ -17,8 +17,9 @@ from dvc.utils import relpath
from dvc.utils.collections import ensure_list
if TYPE_CHECKING:
from scmrepo.base import Base
from dvc.repo import Repo
from dvc.scm.base import Base
logger = logging.getLogger(__name__)
@@ -48,7 +49,7 @@ class SCMContext:
return f"\tgit add {files}".expandtabs(4)
def add(self, paths: Union[str, Iterable[str]]) -> None:
from dvc.scm.exceptions import UnsupportedIndexFormat
from scmrepo.exceptions import UnsupportedIndexFormat
try:
return self.scm.add(paths)
@@ -70,8 +71,9 @@ class SCMContext:
return self.add(self.files_to_track)
def ignore(self, path: str) -> None:
from scmrepo.exceptions import FileNotInRepoError
from dvc.scm import SCMError
from dvc.scm.exceptions import FileNotInRepoError
logger.debug("Adding '%s' to gitignore file.", path)
try:
@@ -84,8 +86,9 @@ class SCMContext:
return self.ignored_paths.append(path)
def ignore_remove(self, path: str) -> None:
from scmrepo.exceptions import FileNotInRepoError
from dvc.scm import SCMError
from dvc.scm.exceptions import FileNotInRepoError
logger.debug("Removing '%s' from gitignore file.", path)
try:
+8 -7
View File
@@ -2,14 +2,15 @@
from contextlib import contextmanager
from typing import TYPE_CHECKING, Iterator
from scmrepo.base import Base # noqa: F401, pylint: disable=unused-import
from scmrepo.git import Git
from scmrepo.noscm import NoSCM
from dvc.exceptions import DvcException
from dvc.progress import Tqdm
from dvc.scm.base import Base # noqa: F401
from dvc.scm.git import Git
from dvc.scm.noscm import NoSCM
if TYPE_CHECKING:
from dvc.scm.progress import GitProgressEvent
from scmrepo.progress import GitProgressEvent
class SCMError(DvcException):
@@ -46,7 +47,7 @@ class GitAuthError(SCMError):
@contextmanager
def map_scm_exception(with_cause: bool = False) -> Iterator[None]:
from dvc.scm.exceptions import SCMError as InternalSCMError
from scmrepo.exceptions import SCMError as InternalSCMError
try:
yield
@@ -96,7 +97,7 @@ class TqdmGit(Tqdm):
def clone(url: str, to_path: str, **kwargs):
from dvc.scm.exceptions import CloneError as InternalCloneError
from scmrepo.exceptions import CloneError as InternalCloneError
with TqdmGit(desc="Cloning") as pbar:
try:
@@ -106,7 +107,7 @@ def clone(url: str, to_path: str, **kwargs):
def resolve_rev(scm: "Git", rev: str) -> str:
from dvc.scm.exceptions import RevError as InternalRevError
from scmrepo.exceptions import RevError as InternalRevError
try:
return scm.resolve_rev(rev)
-48
View File
@@ -1,48 +0,0 @@
"""DVC re-implementation of fsspec's dedicated async event loop."""
import asyncio
import os
import threading
from typing import List, Optional
from fsspec.asyn import ( # noqa: F401, pylint:disable=unused-import
_selector_policy,
sync,
sync_wrapper,
)
# dedicated async IO thread
iothread: List[Optional[threading.Thread]] = [None]
# global DVC event loop
default_loop: List[Optional[asyncio.AbstractEventLoop]] = [None]
lock = threading.Lock()
def get_loop() -> asyncio.AbstractEventLoop:
"""Create or return the global DVC event loop."""
if default_loop[0] is None:
with lock:
if default_loop[0] is None:
with _selector_policy():
default_loop[0] = asyncio.new_event_loop()
loop = default_loop[0]
th = threading.Thread(
target=loop.run_forever, # type: ignore[attr-defined]
name="dvcIO",
)
th.daemon = True
th.start()
iothread[0] = th
assert default_loop[0] is not None
return default_loop[0]
class BaseAsyncObject:
def __init__(self, loop: Optional[asyncio.AbstractEventLoop] = None):
self._loop: asyncio.AbstractEventLoop = loop or get_loop()
self._pid = os.getpid()
@property
def loop(self):
# AsyncMixin is not fork-safe
assert self._pid == os.getpid()
return self._loop
-102
View File
@@ -1,102 +0,0 @@
"""Manages source control systems (e.g. Git) in DVC."""
class Base:
"""Base class for source control management driver implementations."""
def __init__(self, root_dir=None):
import os
self._root_dir = os.path.realpath(root_dir or os.curdir)
@property
def root_dir(self) -> str:
return self._root_dir
def __repr__(self):
return "{class_name}: '{directory}'".format(
class_name=type(self).__name__, directory=self.dir
)
@property
def dir(self):
"""Path to a directory with SCM specific information."""
return None
@staticmethod
def is_repo(root_dir): # pylint: disable=unused-argument
"""Returns whether or not root_dir is a valid SCM repository."""
return True
@staticmethod
def is_submodule(root_dir): # pylint: disable=unused-argument
"""Returns whether or not root_dir is a valid SCM repository
submodule.
"""
return True
def is_ignored(self, path): # pylint: disable=unused-argument
"""Returns whether or not path is ignored by SCM."""
return False
def ignore(self, path): # pylint: disable=unused-argument
"""Makes SCM ignore a specified path."""
def ignore_remove(self, path): # pylint: disable=unused-argument
"""Makes SCM stop ignoring a specified path."""
@property
def ignore_file(self):
"""Filename for a file that contains ignored paths for this SCM."""
def add(self, paths):
"""Makes SCM track every path from a specified list of paths."""
def commit(self, msg):
"""Makes SCM create a commit."""
def checkout(self, branch, create_new=False):
"""Makes SCM checkout a branch."""
def branch(self, branch):
"""Makes SCM create a branch with a specified name."""
def tag(self, tag):
"""Makes SCM create a tag with a specified name."""
def untracked_files(self): # pylint: disable=no-self-use
"""Returns a list of untracked files."""
return []
def is_tracked(self, path): # pylint: disable=no-self-use, unused-argument
"""Returns whether or not a specified path is tracked."""
return False
def is_dirty(self):
"""Return whether the SCM contains uncommitted changes."""
return False
def active_branch(self): # pylint: disable=no-self-use
"""Returns current branch in the repo."""
return ""
def list_branches(self): # pylint: disable=no-self-use
"""Returns a list of available branches in the repo."""
return []
def list_tags(self): # pylint: disable=no-self-use
"""Returns a list of available tags in the repo."""
return []
def list_all_commits(self): # pylint: disable=no-self-use
"""Returns a list of commits in the repo."""
return []
def belongs_to_scm(self, path):
"""Return boolean whether file belongs to scm"""
def close(self):
"""Method to close the files"""
def _reset(self) -> None:
pass
-45
View File
@@ -1,45 +0,0 @@
class SCMError(Exception):
"""Base class for source control management errors."""
class GitHookAlreadyExists(SCMError):
def __init__(self, name: str) -> None:
self.name = name
super().__init__(f"Hook '{name}' already exists")
class FileNotInRepoError(SCMError):
"""Thrown when trying to find .gitignore for a file that is not in a scm
repository.
"""
class MergeConflictError(SCMError):
pass
class InvalidRemote(SCMError):
def __init__(self, url: str) -> None:
self.url = url
super().__init__(f"'{url}' is not a valid Git remote or URL")
class AuthError(SCMError):
def __init__(self, url: str) -> None:
self.url = url
super().__init__(f"HTTP Git authentication is not supported: '{url}'")
class CloneError(SCMError):
def __init__(self, url: str, path: str) -> None:
self.url = url
self.path = path
super().__init__(f"Failed to clone repo '{url}' to '{path}'")
class RevError(SCMError):
pass
class UnsupportedIndexFormat(SCMError):
pass
-376
View File
@@ -1,376 +0,0 @@
"""Manages Git."""
import logging
import os
import re
from collections.abc import Mapping
from contextlib import contextmanager
from functools import partialmethod
from typing import Dict, Iterable, Optional, Tuple, Type
from funcy import cached_property, first
from pathspec.patterns import GitWildMatchPattern
from dvc.scm.base import Base
from dvc.scm.exceptions import (
FileNotInRepoError,
GitHookAlreadyExists,
RevError,
)
from dvc.scm.utils import relpath
from .backend.base import BaseGitBackend, NoGitBackendError
from .backend.dulwich import DulwichBackend
from .backend.gitpython import GitPythonBackend
from .backend.pygit2 import Pygit2Backend
from .stash import Stash
logger = logging.getLogger(__name__)
BackendCls = Type[BaseGitBackend]
class GitBackends(Mapping):
DEFAULT: Dict[str, BackendCls] = {
"dulwich": DulwichBackend,
"pygit2": Pygit2Backend,
"gitpython": GitPythonBackend,
}
def __getitem__(self, key: str) -> BaseGitBackend:
"""Lazily initialize backends and cache it afterwards"""
initialized = self.initialized.get(key)
if not initialized:
backend = self.backends[key]
initialized = backend(*self.args, **self.kwargs)
self.initialized[key] = initialized
return initialized
def __init__(
self, selected: Optional[Iterable[str]], *args, **kwargs
) -> None:
selected = selected or list(self.DEFAULT)
self.backends = {key: self.DEFAULT[key] for key in selected}
self.initialized: Dict[str, BaseGitBackend] = {}
self.args = args
self.kwargs = kwargs
def __iter__(self):
return iter(self.backends)
def __len__(self) -> int:
return len(self.backends)
def close_initialized(self) -> None:
for backend in self.initialized.values():
backend.close()
def reset_all(self) -> None:
for backend in self.initialized.values():
backend._reset() # pylint: disable=protected-access
class Git(Base):
"""Class for managing Git."""
GITIGNORE = ".gitignore"
GIT_DIR = ".git"
LOCAL_BRANCH_PREFIX = "refs/heads/"
RE_HEXSHA = re.compile(r"^[0-9A-Fa-f]{4,40}$")
BAD_REF_CHARS_RE = re.compile("[\177\\s~^:?*\\[]")
def __init__(
self, *args, backends: Optional[Iterable[str]] = None, **kwargs
):
self.backends = GitBackends(backends, *args, **kwargs)
first_ = first(self.backends.values())
super().__init__(first_.root_dir)
@property
def dir(self):
return first(self.backends.values()).dir
@cached_property
def hooks_dir(self):
from pathlib import Path
return Path(self.root_dir) / self.GIT_DIR / "hooks"
@property
def gitpython(self):
return self.backends["gitpython"]
@property
def dulwich(self):
return self.backends["dulwich"]
@property
def pygit2(self):
return self.backends["pygit2"]
@cached_property
def stash(self):
return Stash(self)
@classmethod
def clone(cls, url, to_path, **kwargs):
for _, backend in GitBackends.DEFAULT.items():
try:
backend.clone(url, to_path, **kwargs)
return Git(to_path)
except NotImplementedError:
pass
raise NoGitBackendError("clone")
@classmethod
def is_sha(cls, rev):
return rev and cls.RE_HEXSHA.search(rev)
@classmethod
def split_ref_pattern(cls, ref: str) -> Tuple[str, str]:
name = cls.BAD_REF_CHARS_RE.split(ref, maxsplit=1)[0]
return name, ref[len(name) :]
@staticmethod
def _get_git_dir(root_dir):
return os.path.join(root_dir, Git.GIT_DIR)
@property
def ignore_file(self):
return self.GITIGNORE
def _get_gitignore(self, path):
ignore_file_dir = os.path.dirname(path)
assert os.path.isabs(path)
assert os.path.isabs(ignore_file_dir)
entry = relpath(path, ignore_file_dir).replace(os.sep, "/")
# NOTE: using '/' prefix to make path unambiguous
if len(entry) > 0 and entry[0] != "/":
entry = "/" + entry
gitignore = os.path.join(ignore_file_dir, self.GITIGNORE)
if not os.path.realpath(gitignore).startswith(self.root_dir + os.sep):
raise FileNotInRepoError(
f"'{path}' is outside of git repository '{self.root_dir}'"
)
return entry, gitignore
def ignore(self, path: str) -> Optional[str]:
entry, gitignore = self._get_gitignore(path)
if self.is_ignored(path):
return None
self._add_entry_to_gitignore(entry, gitignore)
return gitignore
def _add_entry_to_gitignore(self, entry, gitignore):
entry = GitWildMatchPattern.escape(entry)
with open(gitignore, "a+", encoding="utf-8") as fobj:
unique_lines = set(fobj.readlines())
fobj.seek(0, os.SEEK_END)
if fobj.tell() == 0:
# Empty file
prefix = ""
else:
fobj.seek(fobj.tell() - 1, os.SEEK_SET)
last = fobj.read(1)
prefix = "" if last == "\n" else "\n"
new_entry = f"{prefix}{entry}\n"
if new_entry not in unique_lines:
fobj.write(new_entry)
def ignore_remove(self, path: str) -> Optional[str]:
entry, gitignore = self._get_gitignore(path)
if not os.path.exists(gitignore):
return None
with open(gitignore, encoding="utf-8") as fobj:
lines = fobj.readlines()
filtered = list(filter(lambda x: x.strip() != entry.strip(), lines))
if not filtered:
os.unlink(gitignore)
return None
with open(gitignore, "w", encoding="utf-8") as fobj:
fobj.writelines(filtered)
return gitignore
def verify_hook(self, name):
if (self.hooks_dir / name).exists():
raise GitHookAlreadyExists(name)
def install_hook(
self, name: str, script: str, interpreter: str = "sh"
) -> None:
import shutil
self.hooks_dir.mkdir(exist_ok=True)
hook = self.hooks_dir / name
directive = f"#!{shutil.which(interpreter)}"
hook.write_text(f"{directive}\n{script}\n", encoding="utf-8")
hook.chmod(0o777)
def install_merge_driver(
self, name: str, description: str, driver: str
) -> None:
self.gitpython.repo.git.config(f"merge.{name}.name", description)
self.gitpython.repo.git.config(f"merge.{name}.driver", driver)
def belongs_to_scm(self, path):
basename = os.path.basename(path)
path_parts = os.path.normpath(path).split(os.path.sep)
return basename == self.ignore_file or Git.GIT_DIR in path_parts
def has_rev(self, rev):
try:
self.resolve_rev(rev)
return True
except RevError:
return False
def close(self):
self.backends.close_initialized()
@property
def no_commits(self):
return not bool(self.get_ref("HEAD"))
def _backend_func(self, name, *args, **kwargs):
for backend in self.backends.values():
try:
func = getattr(backend, name)
return func(*args, **kwargs)
except NotImplementedError:
pass
raise NoGitBackendError(name)
def get_fs(self, rev: str):
from dvc.fs.git import GitFileSystem
return GitFileSystem(scm=self, rev=rev)
is_ignored = partialmethod(_backend_func, "is_ignored")
add = partialmethod(_backend_func, "add")
commit = partialmethod(_backend_func, "commit")
checkout = partialmethod(_backend_func, "checkout")
pull = partialmethod(_backend_func, "pull")
push = partialmethod(_backend_func, "push")
branch = partialmethod(_backend_func, "branch")
tag = partialmethod(_backend_func, "tag")
untracked_files = partialmethod(_backend_func, "untracked_files")
is_tracked = partialmethod(_backend_func, "is_tracked")
is_dirty = partialmethod(_backend_func, "is_dirty")
active_branch = partialmethod(_backend_func, "active_branch")
list_branches = partialmethod(_backend_func, "list_branches")
list_tags = partialmethod(_backend_func, "list_tags")
list_all_commits = partialmethod(_backend_func, "list_all_commits")
get_rev = partialmethod(_backend_func, "get_rev")
resolve_rev = partialmethod(_backend_func, "resolve_rev")
resolve_commit = partialmethod(_backend_func, "resolve_commit")
set_ref = partialmethod(_backend_func, "set_ref")
get_ref = partialmethod(_backend_func, "get_ref")
remove_ref = partialmethod(_backend_func, "remove_ref")
iter_refs = partialmethod(_backend_func, "iter_refs")
iter_remote_refs = partialmethod(_backend_func, "iter_remote_refs")
get_refs_containing = partialmethod(_backend_func, "get_refs_containing")
push_refspec = partialmethod(_backend_func, "push_refspec")
fetch_refspecs = partialmethod(_backend_func, "fetch_refspecs")
_stash_iter = partialmethod(_backend_func, "_stash_iter")
_stash_push = partialmethod(_backend_func, "_stash_push")
_stash_apply = partialmethod(_backend_func, "_stash_apply")
_stash_drop = partialmethod(_backend_func, "_stash_drop")
describe = partialmethod(_backend_func, "describe")
diff = partialmethod(_backend_func, "diff")
reset = partialmethod(_backend_func, "reset")
checkout_index = partialmethod(_backend_func, "checkout_index")
status = partialmethod(_backend_func, "status")
merge = partialmethod(_backend_func, "merge")
validate_git_remote = partialmethod(_backend_func, "validate_git_remote")
check_ref_format = partialmethod(_backend_func, "check_ref_format")
def branch_revs(
self, branch: str, end_rev: Optional[str] = None
) -> Iterable[str]:
"""Iterate over revisions in a given branch (from newest to oldest).
If end_rev is set, iterator will stop when the specified revision is
reached.
"""
commit = self.resolve_commit(branch)
while commit is not None:
yield commit.hexsha
parent = first(commit.parents)
if parent is None or parent == end_rev:
return
commit = self.resolve_commit(parent)
@contextmanager
def detach_head(
self,
rev: Optional[str] = None,
force: bool = False,
client: str = "scm",
):
"""Context manager for performing detached HEAD SCM operations.
Detaches and restores HEAD similar to interactive git rebase.
Restore is equivalent to 'reset --soft', meaning the caller is
is responsible for preserving & restoring working tree state
(i.e. via stash) when applicable.
Yields revision of detached head.
"""
if not rev:
rev = "HEAD"
orig_head = self.get_ref("HEAD", follow=False)
logger.debug("Detaching HEAD at '%s'", rev)
self.checkout(rev, detach=True, force=force)
try:
yield self.get_ref("HEAD")
finally:
prefix = self.LOCAL_BRANCH_PREFIX
if orig_head.startswith(prefix):
symbolic = True
name = orig_head[len(prefix) :]
else:
symbolic = False
name = orig_head
self.set_ref(
"HEAD",
orig_head,
symbolic=symbolic,
message=f"{client}: Restore HEAD to '{name}'",
)
logger.debug("Restore HEAD to '%s'", name)
self.reset()
@contextmanager
def stash_workspace(self, **kwargs):
"""Stash and restore any workspace changes.
Yields revision of the stash commit. Yields None if there were no
changes to stash.
"""
logger.debug("Stashing workspace")
rev = self.stash.push(**kwargs)
try:
yield rev
finally:
if rev:
logger.debug("Restoring stashed workspace")
self.stash.pop()
def _reset(self) -> None:
self.backends.reset_all()
View File
-360
View File
@@ -1,360 +0,0 @@
import os
from abc import ABC, abstractmethod
from typing import (
TYPE_CHECKING,
Callable,
Iterable,
Mapping,
Optional,
Tuple,
Union,
)
from dvc.scm.exceptions import SCMError
from ..objects import GitObject
if TYPE_CHECKING:
from dvc.scm.progress import GitProgressEvent
from ..objects import GitCommit
class NoGitBackendError(SCMError):
def __init__(self, func):
super().__init__(f"No valid Git backend for '{func}'")
class BaseGitBackend(ABC):
"""Base Git backend class."""
@abstractmethod
def __init__(self, root_dir=os.curdir, search_parent_directories=True):
pass
def close(self):
pass
@abstractmethod
def is_ignored(self, path: str) -> bool:
"""Return True if the specified path is gitignored."""
@property
@abstractmethod
def root_dir(self) -> str:
pass
@staticmethod
@abstractmethod
def clone(
url: str,
to_path: str,
rev: Optional[str] = None,
shallow_branch: Optional[str] = None,
progress: Callable[["GitProgressEvent"], None] = None,
):
pass
@property
@abstractmethod
def dir(self) -> str:
pass
@abstractmethod
def add(self, paths: Union[str, Iterable[str]], update=False):
pass
@abstractmethod
def commit(self, msg: str, no_verify: bool = False):
pass
@abstractmethod
def checkout(
self,
branch: str,
create_new: Optional[bool] = False,
force: bool = False,
**kwargs,
):
pass
@abstractmethod
def pull(self, **kwargs):
pass
@abstractmethod
def push(self):
pass
@abstractmethod
def branch(self, branch: str):
pass
@abstractmethod
def tag(self, tag: str):
pass
@abstractmethod
def untracked_files(self) -> Iterable[str]:
pass
@abstractmethod
def is_tracked(self, path: str) -> bool:
pass
@abstractmethod
def is_dirty(self, untracked_files: bool = False) -> bool:
pass
@abstractmethod
def active_branch(self) -> str:
pass
@abstractmethod
def list_branches(self) -> Iterable[str]:
pass
@abstractmethod
def list_tags(self) -> Iterable[str]:
pass
@abstractmethod
def list_all_commits(self) -> Iterable[str]:
pass
@abstractmethod
def get_tree_obj(self, rev: str, **kwargs) -> GitObject:
pass
@abstractmethod
def get_rev(self) -> str:
pass
@abstractmethod
def resolve_rev(self, rev: str) -> str:
pass
@abstractmethod
def resolve_commit(self, rev: str) -> "GitCommit":
pass
@abstractmethod
def set_ref(
self,
name: str,
new_ref: str,
old_ref: Optional[str] = None,
message: Optional[str] = None,
symbolic: Optional[bool] = False,
):
"""Set the specified git ref.
Optional kwargs:
old_ref: If specified, ref will only be set if it currently equals
old_ref. Has no effect is symbolic is True.
message: Optional reflog message.
symbolic: If True, ref will be set as a symbolic ref to new_ref
rather than the dereferenced object.
"""
@abstractmethod
def get_ref(self, name: str, follow: bool = True) -> Optional[str]:
"""Return the value of specified ref.
If follow is false, symbolic refs will not be dereferenced.
Returns None if the ref does not exist.
"""
@abstractmethod
def remove_ref(self, name: str, old_ref: Optional[str] = None):
"""Remove the specified git ref.
If old_ref is specified, ref will only be removed if it currently
equals old_ref.
"""
@abstractmethod
def iter_refs(self, base: Optional[str] = None):
"""Iterate over all refs in this git repo.
If base is specified, only refs which begin with base will be yielded.
"""
@abstractmethod
def iter_remote_refs(self, url: str, base: Optional[str] = None, **kwargs):
"""Iterate over all refs in the specified remote Git repo.
If base is specified, only refs which begin with base will be yielded.
URL can be a named Git remote or URL.
"""
@abstractmethod
def get_refs_containing(self, rev: str, pattern: Optional[str] = None):
"""Iterate over all git refs containing the specified revision."""
@abstractmethod
def push_refspec(
self,
url: str,
src: Optional[str],
dest: str,
force: bool = False,
on_diverged: Optional[Callable[[str, str], bool]] = None,
progress: Callable[["GitProgressEvent"], None] = None,
**kwargs,
):
"""Push refspec to a remote Git repo.
Args:
url: Git remote name or absolute Git URL.
src: Local refspec. If src ends with "/" it will be treated as a
prefix, and all refs inside src will be pushed using dest
as destination refspec prefix. If src is None, dest will be
deleted from the remote.
dest: Remote refspec.
force: If True, remote refs will be overwritten.
on_diverged: Callback function which will be called if local ref
and remote have diverged and force is False. If the callback
returns True the remote ref will be overwritten.
Callback will be of the form:
on_diverged(local_refname, remote_sha)
"""
@abstractmethod
def fetch_refspecs(
self,
url: str,
refspecs: Iterable[str],
force: Optional[bool] = False,
on_diverged: Optional[Callable[[str, str], bool]] = None,
progress: Callable[["GitProgressEvent"], None] = None,
**kwargs,
):
"""Fetch refspecs from a remote Git repo.
Args:
url: Remote repo Git URL (Note this must be a Git URL and not
a remote name).
refspecs: Iterable containing refspecs to fetch.
Note that this will not match subkeys.
force: If True, local refs will be overwritten.
on_diverged: Callback function which will be called if local ref
and remote have diverged and force is False. If the callback
returns True the local ref will be overwritten.
Callback will be of the form:
on_diverged(local_refname, remote_sha)
"""
@abstractmethod
def _stash_iter(self, ref: str):
"""Iterate over stash commits in the specified ref."""
@abstractmethod
def _stash_push(
self,
ref: str,
message: Optional[str] = None,
include_untracked: Optional[bool] = False,
) -> Tuple[Optional[str], bool]:
"""Push a commit onto the specified stash.
Returns a tuple of the form (rev, need_reset) where need_reset
indicates whether or not the workspace should be `reset --hard`
(some backends will not clean the workspace after creating a stash
commit).
"""
@abstractmethod
def _stash_apply(self, rev: str):
"""Apply the specified stash revision."""
@abstractmethod
def _stash_drop(self, ref: str, index: int):
"""Drop the specified stash revision."""
@abstractmethod
def describe(
self,
rev: str,
base: Optional[str] = None,
match: Optional[str] = None,
exclude: Optional[str] = None,
) -> Optional[str]:
"""Return the first ref which points to rev.
Roughly equivalent to `git describe --all --exact-match`.
If no matching ref was found, returns None.
Optional kwargs:
base: Base ref prefix to search, defaults to "refs/tags"
match: Glob pattern. If specified only refs matching this glob
pattern will be returned.
exclude: Glob pattern. If specified, only refs which do not match
this pattern will be returned.
"""
@abstractmethod
def diff(self, rev_a: str, rev_b: str, binary=False) -> str:
"""Return the git diff for two commits."""
@abstractmethod
def reset(self, hard: bool = False, paths: Iterable[str] = None):
"""Reset current git HEAD."""
@abstractmethod
def checkout_index(
self,
paths: Optional[Iterable[str]] = None,
force: bool = False,
ours: bool = False,
theirs: bool = False,
):
"""Checkout the specified paths from index."""
@abstractmethod
def status(
self, ignored: bool = False
) -> Tuple[Mapping[str, Iterable[str]], Iterable[str], Iterable[str]]:
"""Return tuple of (staged_files, unstaged_files, untracked_files).
staged_files will be a dict mapping status (add, delete, modify) to a
list of paths.
If ignored is True, gitignored files will be included in
untracked_paths.
"""
def _reset(self) -> None:
pass
@abstractmethod
def merge(
self,
rev: str,
commit: bool = True,
msg: Optional[str] = None,
squash: bool = False,
) -> Optional[str]:
"""Merge specified commit into the current (working copy) branch.
Args:
rev: Git revision to merge.
commit: If True, merge will be committed using `msg` as the commit
message. If False, the merge changes will be staged but not
committed.
msg: Merge commit message, required if `commit` is True.
squash: If True, working tree state will be updated with merge
results, but no commit will be made and no changes will be
staged. `commit` must be False when `squash` is True.
If conflicts are present after merging, MergeConflictError will be
raised. The caller is responsible for either resolving the conflicts
or resetting the repository state.
Returns revision of the merge commit or None if no commit was made.
"""
@abstractmethod
def validate_git_remote(self, url: str, **kwargs):
"""Verify that url is a valid git URL or remote name."""
-676
View File
@@ -1,676 +0,0 @@
import fnmatch
import locale
import logging
import os
import stat
from io import BytesIO, StringIO
from typing import (
TYPE_CHECKING,
Callable,
Dict,
Iterable,
List,
Mapping,
Optional,
Tuple,
Union,
)
from funcy import cached_property
from dvc.scm.exceptions import AuthError, InvalidRemote, SCMError
from dvc.scm.utils import relpath
from ...objects import GitObject
from ..base import BaseGitBackend
if TYPE_CHECKING:
from dvc.scm.progress import GitProgressEvent
from ...objects import GitCommit
logger = logging.getLogger(__name__)
class DulwichObject(GitObject):
def __init__(self, repo, name, mode, sha):
self.repo = repo
self._name = name
self._mode = mode
self._sha = sha
def open(self, mode: str = "r", encoding: str = None):
if not encoding:
encoding = locale.getpreferredencoding(False)
# NOTE: we didn't load the object before as Dulwich will also try to
# load the contents of it into memory, which will slow down Trie
# building considerably.
obj = self.repo[self.sha]
data = obj.as_raw_string()
if mode == "rb":
return BytesIO(data)
return StringIO(data.decode(encoding))
@property
def name(self) -> str:
return self._name
@property
def mode(self) -> int:
return self._mode
def scandir(self) -> Iterable["DulwichObject"]:
tree = self.repo[self.sha]
for entry in tree.iteritems(): # noqa: B301
yield DulwichObject(
self.repo, entry.path.decode(), entry.mode, entry.sha
)
@property
def size(self) -> int:
return len(self.repo[self.sha].as_raw_string())
@property
def sha(self) -> str:
return self._sha
class DulwichBackend(BaseGitBackend): # pylint:disable=abstract-method
"""Dulwich Git backend."""
from dulwich import client
from .asyncssh_vendor import AsyncSSHVendor
# monkeypatch dulwich client's default SSH vendor to use asyncssh
client.get_ssh_vendor = AsyncSSHVendor # type: ignore[assignment]
# Dulwich progress will return messages equivalent to git CLI,
# our pbars should just display the messages as formatted by dulwich
BAR_FMT_NOTOTAL = "{desc}{bar:b}|{postfix[info]} [{elapsed}]"
def __init__( # pylint:disable=W0231
self, root_dir=os.curdir, search_parent_directories=True
):
from dulwich.errors import NotGitRepository
from dulwich.repo import Repo
try:
if search_parent_directories:
self.repo = Repo.discover(start=root_dir)
else:
self.repo = Repo(root_dir)
except NotGitRepository as exc:
raise SCMError(f"{root_dir} is not a git repository") from exc
self._submodules: Dict[str, str] = self._find_submodules()
self._stashes: dict = {}
def _find_submodules(self) -> Dict[str, str]:
"""Return dict mapping submodule names to submodule paths.
Submodule paths will be relative to Git repo root.
"""
from dulwich.config import ConfigFile, parse_submodules
submodules: Dict[str, str] = {}
config_path = os.path.join(self.root_dir, ".gitmodules")
if os.path.isfile(config_path):
config = ConfigFile.from_path(config_path)
for path, _url, section in parse_submodules(config):
submodules[os.fsdecode(section)] = os.fsdecode(path)
return submodules
def close(self):
self.repo.close()
@property
def root_dir(self) -> str:
return self.repo.path
@staticmethod
def clone(
url: str,
to_path: str,
rev: Optional[str] = None,
shallow_branch: Optional[str] = None,
progress: Callable[["GitProgressEvent"], None] = None,
):
raise NotImplementedError
@property
def dir(self) -> str:
return self.repo.commondir()
def add(self, paths: Union[str, Iterable[str]], update=False):
assert paths or update
if isinstance(paths, str):
paths = [paths]
if update and not paths:
self.repo.stage(list(self.repo.open_index()))
return
files: List[bytes] = []
for path in paths:
if not os.path.isabs(path) and self._submodules:
# NOTE: If path is inside a submodule, Dulwich expects the
# staged paths to be relative to the submodule root (not the
# parent git repo root). We append path to root_dir here so
# that the result of relpath(path, root_dir) is actually the
# path relative to the submodule root.
fs_path = relpath(path, self.root_dir)
for sm_path in self._submodules.values():
if fs_path.startswith(sm_path):
path = os.path.join(
self.root_dir,
relpath(fs_path, sm_path),
)
break
if os.path.isdir(path):
files.extend(
os.fsencode(
relpath(os.path.join(root, fpath), self.root_dir)
)
for root, _, fs in os.walk(path)
for fpath in fs
)
else:
files.append(os.fsencode(relpath(path, self.root_dir)))
# NOTE: this doesn't check gitignore, same as GitPythonBackend.add
if update:
index = self.repo.open_index()
if os.name == "nt":
# NOTE: we need git/unix separator to compare against index
# paths but repo.stage() expects to be called with OS paths
self.repo.stage(
[
fname
for fname in files
if fname.replace(b"\\", b"/") in index
]
)
else:
self.repo.stage([fname for fname in files if fname in index])
else:
self.repo.stage(files)
def commit(self, msg: str, no_verify: bool = False):
from dulwich.errors import CommitError
from dulwich.porcelain import commit
from dulwich.repo import InvalidUserIdentity
try:
commit(self.root_dir, message=msg, no_verify=no_verify)
except CommitError as exc:
raise SCMError("Git commit failed") from exc
except InvalidUserIdentity as exc:
raise SCMError(
"Git username and email must be configured"
) from exc
def checkout(
self,
branch: str,
create_new: Optional[bool] = False,
force: bool = False,
**kwargs,
):
raise NotImplementedError
def pull(self, **kwargs):
raise NotImplementedError
def push(self):
raise NotImplementedError
def branch(self, branch: str):
from dulwich.porcelain import Error, branch_create
try:
branch_create(self.root_dir, branch)
except Error as exc:
raise SCMError(f"Failed to create branch '{branch}'") from exc
def tag(self, tag: str):
raise NotImplementedError
def untracked_files(self) -> Iterable[str]:
_staged, _unstaged, untracked = self.status()
return untracked
def is_tracked(self, path: str) -> bool:
rel = relpath(path, self.root_dir).replace(os.path.sep, "/").encode()
rel_dir = rel + b"/"
for p in self.repo.open_index():
if p == rel or p.startswith(rel_dir):
return True
return False
def is_dirty(self, untracked_files: bool = False) -> bool:
staged, unstaged, untracked = self.status()
return bool(staged or unstaged or (untracked_files and untracked))
def active_branch(self) -> str:
raise NotImplementedError
def list_branches(self) -> Iterable[str]:
raise NotImplementedError
def list_tags(self) -> Iterable[str]:
raise NotImplementedError
def list_all_commits(self) -> Iterable[str]:
raise NotImplementedError
def get_tree_obj(self, rev: str, **kwargs) -> DulwichObject:
from dulwich.objectspec import parse_tree
tree = parse_tree(self.repo, rev)
return DulwichObject(self.repo, ".", stat.S_IFDIR, tree.id)
def get_rev(self) -> str:
rev = self.get_ref("HEAD")
if rev:
return rev
raise SCMError("Empty git repo")
def resolve_rev(self, rev: str) -> str:
raise NotImplementedError
def resolve_commit(self, rev: str) -> "GitCommit":
raise NotImplementedError
def _get_stash(self, ref: str):
from dulwich.stash import Stash as DulwichStash
if ref not in self._stashes:
self._stashes[ref] = DulwichStash(self.repo, ref=os.fsencode(ref))
return self._stashes[ref]
@cached_property
def ignore_manager(self):
from dulwich.ignore import IgnoreFilterManager
return IgnoreFilterManager.from_repo(self.repo)
def is_ignored(self, path: "Union[str, os.PathLike[str]]") -> bool:
# `is_ignored` returns `false` if excluded in `.gitignore` and
# `None` if it's not mentioned at all. `True` if it is ignored.
relative_path = relpath(path, self.root_dir)
# if checking a directory, a trailing slash must be included
if str(path)[-1] == os.sep:
relative_path += os.sep
return bool(self.ignore_manager.is_ignored(relative_path))
def set_ref(
self,
name: str,
new_ref: str,
old_ref: Optional[str] = None,
message: Optional[str] = None,
symbolic: Optional[bool] = False,
):
name_b = os.fsencode(name)
new_ref_b = os.fsencode(new_ref)
old_ref_b = os.fsencode(old_ref) if old_ref else None
message_b = message.encode("utf-8") if message else None
if symbolic:
return self.repo.refs.set_symbolic_ref(
name_b, new_ref_b, message=message_b
)
if not self.repo.refs.set_if_equals(
name_b, old_ref_b, new_ref_b, message=message_b
):
raise SCMError(f"Failed to set '{name}'")
def get_ref(self, name, follow: bool = True) -> Optional[str]:
from dulwich.refs import parse_symref_value
name_b = os.fsencode(name)
if follow:
try:
ref = self.repo.refs[name_b]
except KeyError:
ref = None
else:
ref = self.repo.refs.read_ref(name_b)
try:
if ref:
ref = parse_symref_value(ref)
except ValueError:
pass
if ref:
return os.fsdecode(ref)
return None
def remove_ref(self, name: str, old_ref: Optional[str] = None):
name_b = name.encode("utf-8")
old_ref_b = old_ref.encode("utf-8") if old_ref else None
if not self.repo.refs.remove_if_equals(name_b, old_ref_b):
raise SCMError(f"Failed to remove '{name}'")
def iter_refs(self, base: Optional[str] = None):
base_b = os.fsencode(base) if base else None
for key in self.repo.refs.keys(base=base_b):
if base:
if base.endswith("/"):
base = base[:-1]
yield "/".join([base, os.fsdecode(key)])
else:
yield os.fsdecode(key)
def iter_remote_refs(self, url: str, base: Optional[str] = None, **kwargs):
from dulwich.client import HTTPUnauthorized, get_transport_and_path
from dulwich.errors import NotGitRepository
from dulwich.porcelain import get_remote_repo
try:
_remote, location = get_remote_repo(self.repo, url)
client, path = get_transport_and_path(location, **kwargs)
except Exception as exc:
raise InvalidRemote(url) from exc
try:
if base:
yield from (
os.fsdecode(ref)
for ref in client.get_refs(path)
if ref.startswith(os.fsencode(base))
)
else:
yield from (os.fsdecode(ref) for ref in client.get_refs(path))
except NotGitRepository as exc:
raise InvalidRemote(url) from exc
except HTTPUnauthorized:
raise AuthError(url)
def get_refs_containing(self, rev: str, pattern: Optional[str] = None):
raise NotImplementedError
def push_refspec(
self,
url: str,
src: Optional[str],
dest: str,
force: bool = False,
on_diverged: Optional[Callable[[str, str], bool]] = None,
progress: Callable[["GitProgressEvent"], None] = None,
**kwargs,
):
from dulwich.client import HTTPUnauthorized, get_transport_and_path
from dulwich.errors import NotGitRepository, SendPackError
from dulwich.porcelain import (
DivergedBranches,
check_diverged,
get_remote_repo,
)
dest_refs, values = self._push_dest_refs(src, dest)
try:
_remote, location = get_remote_repo(self.repo, url)
client, path = get_transport_and_path(location, **kwargs)
except Exception as exc:
raise SCMError(
f"'{url}' is not a valid Git remote or URL"
) from exc
def update_refs(refs):
from dulwich.objects import ZERO_SHA
new_refs = {}
for ref, value in zip(dest_refs, values):
if ref in refs and value != ZERO_SHA:
local_sha = self.repo.refs[ref]
remote_sha = refs[ref]
try:
check_diverged(self.repo, remote_sha, local_sha)
except DivergedBranches:
if not force:
overwrite = False
if on_diverged:
overwrite = on_diverged(
os.fsdecode(ref), os.fsdecode(remote_sha)
)
if not overwrite:
continue
new_refs[ref] = value
return new_refs
try:
from dvc.scm.progress import GitProgressReporter
client.send_pack(
path,
update_refs,
self.repo.object_store.generate_pack_data,
progress=GitProgressReporter(progress) if progress else None,
)
except (NotGitRepository, SendPackError) as exc:
raise SCMError("Git failed to push '{src}' to '{url}'") from exc
except HTTPUnauthorized:
raise AuthError(url)
def _push_dest_refs(
self, src: Optional[str], dest: str
) -> Tuple[Iterable[bytes], Iterable[bytes]]:
from dulwich.objects import ZERO_SHA
if src is not None and src.endswith("/"):
src_b = os.fsencode(src)
keys = self.repo.refs.subkeys(src_b)
values = [self.repo.refs[b"".join([src_b, key])] for key in keys]
dest_refs = [b"".join([os.fsencode(dest), key]) for key in keys]
else:
if src is None:
values = [ZERO_SHA]
else:
values = [self.repo.refs[os.fsencode(src)]]
dest_refs = [os.fsencode(dest)]
return dest_refs, values
def fetch_refspecs(
self,
url: str,
refspecs: Iterable[str],
force: Optional[bool] = False,
on_diverged: Optional[Callable[[str, str], bool]] = None,
progress: Callable[["GitProgressEvent"], None] = None,
**kwargs,
):
from dulwich.client import get_transport_and_path
from dulwich.objectspec import parse_reftuples
from dulwich.porcelain import (
DivergedBranches,
check_diverged,
get_remote_repo,
)
fetch_refs = []
def determine_wants(remote_refs):
fetch_refs.extend(
parse_reftuples(
remote_refs,
self.repo.refs,
[os.fsencode(refspec) for refspec in refspecs],
force=force,
)
)
return [
remote_refs[lh]
for (lh, _, _) in fetch_refs
if remote_refs[lh] not in self.repo.object_store
]
try:
_remote, location = get_remote_repo(self.repo, url)
client, path = get_transport_and_path(location, **kwargs)
except Exception as exc:
raise SCMError(
f"'{url}' is not a valid Git remote or URL"
) from exc
from dvc.scm.progress import GitProgressReporter
fetch_result = client.fetch(
path,
self.repo,
progress=GitProgressReporter(progress) if progress else None,
determine_wants=determine_wants,
)
for (lh, rh, _) in fetch_refs:
try:
if rh in self.repo.refs:
check_diverged(
self.repo, self.repo.refs[rh], fetch_result.refs[lh]
)
except DivergedBranches:
if not force:
overwrite = False
if on_diverged:
overwrite = on_diverged(
os.fsdecode(rh), os.fsdecode(fetch_result.refs[lh])
)
if not overwrite:
continue
self.repo.refs[rh] = fetch_result.refs[lh]
def _stash_iter(self, ref: str):
stash = self._get_stash(ref)
yield from stash.stashes()
def _stash_push(
self,
ref: str,
message: Optional[str] = None,
include_untracked: Optional[bool] = False,
) -> Tuple[Optional[str], bool]:
from dulwich.repo import InvalidUserIdentity
from dvc.scm.git import Stash
if include_untracked or ref == Stash.DEFAULT_STASH:
# dulwich stash.push does not support include_untracked and does
# not touch working tree
raise NotImplementedError
stash = self._get_stash(ref)
message_b = message.encode("utf-8") if message else None
try:
rev = stash.push(message=message_b)
except InvalidUserIdentity as exc:
raise SCMError(
"Git username and email must be configured"
) from exc
return os.fsdecode(rev), True
def _stash_apply(self, rev: str):
raise NotImplementedError
def _stash_drop(self, ref: str, index: int):
from dvc.scm.git import Stash
if ref == Stash.DEFAULT_STASH:
raise NotImplementedError
stash = self._get_stash(ref)
try:
stash.drop(index)
except ValueError as exc:
raise SCMError("Failed to drop stash entry") from exc
def describe(
self,
rev: str,
base: Optional[str] = None,
match: Optional[str] = None,
exclude: Optional[str] = None,
) -> Optional[str]:
if not base:
base = "refs/tags"
for ref in self.iter_refs(base=base):
if (match and not fnmatch.fnmatch(ref, match)) or (
exclude and fnmatch.fnmatch(ref, exclude)
):
continue
if self.get_ref(ref, follow=False) == rev:
return ref
return None
def diff(self, rev_a: str, rev_b: str, binary=False) -> str:
from dulwich.patch import write_tree_diff
commit_a = self.repo[os.fsencode(rev_a)]
commit_b = self.repo[os.fsencode(rev_b)]
buf = BytesIO()
write_tree_diff(
buf, self.repo.object_store, commit_a.tree, commit_b.tree
)
return buf.getvalue().decode("utf-8")
def reset(self, hard: bool = False, paths: Iterable[str] = None):
raise NotImplementedError
def checkout_index(
self,
paths: Optional[Iterable[str]] = None,
force: bool = False,
ours: bool = False,
theirs: bool = False,
):
raise NotImplementedError
def status(
self, ignored: bool = False
) -> Tuple[Mapping[str, Iterable[str]], Iterable[str], Iterable[str]]:
from dulwich.porcelain import status as git_status
staged, unstaged, untracked = git_status(
self.root_dir, ignored=ignored
)
return (
{
status: [os.fsdecode(name) for name in paths]
for status, paths in staged.items()
if paths
},
[os.fsdecode(name) for name in unstaged],
[os.fsdecode(name) for name in untracked],
)
def _reset(self) -> None:
self.__dict__.pop("ignore_manager", None)
def merge(
self,
rev: str,
commit: bool = True,
msg: Optional[str] = None,
squash: bool = False,
) -> Optional[str]:
raise NotImplementedError
def validate_git_remote(self, url: str, **kwargs):
from dulwich.client import LocalGitClient, get_transport_and_path
from dulwich.porcelain import get_remote_repo
try:
_, location = get_remote_repo(self.repo, url)
client, path = get_transport_and_path(location, **kwargs)
except Exception as exc:
raise InvalidRemote(url) from exc
if isinstance(client, LocalGitClient) and not os.path.exists(
os.path.join("", path)
):
raise InvalidRemote(url)
def check_ref_format(self, refname: str):
from dulwich.refs import check_ref_format
return check_ref_format(refname.encode())
@@ -1,144 +0,0 @@
"""asyncssh SSH vendor for Dulwich."""
from typing import List, Optional
from dulwich.client import SSHVendor
from dvc.scm.asyn import BaseAsyncObject, sync_wrapper
class _StderrWrapper:
def __init__(self, stderr):
self.stderr = stderr
async def _readlines(self):
lines = []
while True:
line = await self.stderr.readline()
if not line:
break
lines.append(line)
return lines
readlines = sync_wrapper(_readlines)
class AsyncSSHWrapper(BaseAsyncObject):
def __init__(self, conn, proc, **kwargs):
super().__init__(**kwargs)
self.conn = conn
self.proc = proc
self.stderr = _StderrWrapper(proc.stderr)
def can_read(self) -> bool:
# pylint:disable=protected-access
return self.proc.stdout._session._recv_buf_len > 0
async def _read(self, n: Optional[int] = None) -> bytes:
if self.proc.stdout.at_eof():
return b""
return await self.proc.stdout.read(n=n if n is not None else -1)
read = sync_wrapper(_read)
async def _write(self, data: bytes):
self.proc.stdin.write(data)
await self.proc.stdin.drain()
write = sync_wrapper(_write)
async def _close(self):
self.conn.close()
await self.conn.wait_closed()
close = sync_wrapper(_close)
# NOTE: Github's SSH server does not strictly comply with the SSH protocol.
# When validating a public key using the rsa-sha2-256 or rsa-sha2-512
# signature algorithms, RFC4252 + RFC8332 state that the server should respond
# with the same algorithm in SSH_MSG_USERAUTH_PK_OK. Github's server always
# returns "ssh-rsa" rather than the correct sha2 algorithm name (likely for
# backwards compatibility with old SSH client reasons). This behavior causes
# asyncssh to fail with a key-mismatch error (since asyncssh expects the server
# to behave properly).
#
# See also:
# https://www.ietf.org/rfc/rfc4252.txt
# https://www.ietf.org/rfc/rfc8332.txt
def _process_public_key_ok_gh(self, _pkttype, _pktid, packet):
from asyncssh.misc import ProtocolError
algorithm = packet.get_string()
key_data = packet.get_string()
packet.check_end()
# pylint: disable=protected-access
if (
(
algorithm == b"ssh-rsa"
and self._keypair.algorithm
not in (
b"ssh-rsa",
b"rsa-sha2-256",
b"rsa-sha2-512",
)
)
or (algorithm != b"ssh-rsa" and algorithm != self._keypair.algorithm)
or key_data != self._keypair.public_data
):
raise ProtocolError("Key mismatch")
self.create_task(self._send_signed_request())
return True
class AsyncSSHVendor(BaseAsyncObject, SSHVendor):
def __init__(self, **kwargs):
super().__init__(**kwargs)
async def _run_command(
self,
host: str,
command: List[str],
username: Optional[str] = None,
port: Optional[int] = None,
password: Optional[str] = None,
key_filename: Optional[str] = None,
**kwargs,
):
"""Connect to an SSH server.
Run a command remotely and return a file-like object for interaction
with the remote command.
Args:
host: Host name
command: Command to run (as argv array)
username: Optional ame of user to log in as
port: Optional SSH port to use
password: Optional ssh password for login or private key
key_filename: Optional path to private keyfile
"""
import asyncssh
from asyncssh.auth import MSG_USERAUTH_PK_OK, _ClientPublicKeyAuth
# pylint: disable=protected-access
_ClientPublicKeyAuth._packet_handlers[
MSG_USERAUTH_PK_OK
] = _process_public_key_ok_gh
conn = await asyncssh.connect(
host,
port=port if port is not None else (),
username=username,
password=password,
client_keys=[key_filename] if key_filename else (),
ignore_encrypted=not key_filename,
known_hosts=None,
encoding=None,
)
proc = await conn.create_process(command, encoding=None)
return AsyncSSHWrapper(conn, proc)
run_command = sync_wrapper(_run_command)
-632
View File
@@ -1,632 +0,0 @@
import io
import locale
import logging
import os
import sys
from functools import partial
from typing import (
TYPE_CHECKING,
Callable,
Dict,
Iterable,
List,
Mapping,
Optional,
Tuple,
Union,
)
from funcy import ignore
from dvc.scm.exceptions import (
CloneError,
MergeConflictError,
RevError,
SCMError,
UnsupportedIndexFormat,
)
from dvc.scm.utils import relpath
from ..objects import GitCommit, GitObject
from .base import BaseGitBackend
if TYPE_CHECKING:
from dvc.scm.progress import GitProgressEvent
logger = logging.getLogger(__name__)
# NOTE: Check if we are in a bundle
# https://pythonhosted.org/PyInstaller/runtime-information.html
def is_binary() -> bool:
return getattr(sys, "frozen", False)
def fix_env(env: Dict[str, str] = None) -> Dict[str, str]:
if env is None:
environ = os.environ.copy()
else:
environ = env.copy()
if is_binary():
lp_key = "LD_LIBRARY_PATH"
lp_orig = environ.get(lp_key + "_ORIG", None)
if lp_orig is not None:
environ[lp_key] = lp_orig
else:
environ.pop(lp_key, None)
return environ
class GitPythonObject(GitObject):
def __init__(self, obj):
self.obj = obj
def open(self, mode: str = "r", encoding: str = None):
if not encoding:
encoding = locale.getpreferredencoding(False)
# GitPython's obj.data_stream is a fragile thing, it is better to
# read it immediately, also it needs to be to decoded if we follow
# the `open()` behavior (since data_stream.read() returns bytes,
# and `open` with default "r" mode returns str)
data = self.obj.data_stream.read()
if mode == "rb":
return io.BytesIO(data)
return io.StringIO(data.decode(encoding))
@property
def name(self) -> str:
# NOTE: `obj.name` is not always a basename. See [1] for more details.
#
# [1] https://github.com/iterative/dvc/issues/3481
return os.path.basename(self.obj.path)
@property
def mode(self) -> int:
return self.obj.mode
def scandir(self) -> Iterable["GitPythonObject"]:
for obj in self.obj:
yield GitPythonObject(obj)
@property
def size(self) -> int:
return self.obj.size
@property
def sha(self) -> str:
return self.obj.hexsha
class GitPythonBackend(BaseGitBackend): # pylint:disable=abstract-method
"""git-python Git backend."""
def __init__( # pylint:disable=W0231
self, root_dir=os.curdir, search_parent_directories=True
):
import git
from git.exc import InvalidGitRepositoryError
try:
self.repo = git.Repo(
root_dir, search_parent_directories=search_parent_directories
)
except InvalidGitRepositoryError:
msg = "{} is not a git repository"
raise SCMError(msg.format(root_dir))
# NOTE: fixing LD_LIBRARY_PATH for binary built by PyInstaller.
# http://pyinstaller.readthedocs.io/en/stable/runtime-information.html
env = fix_env()
libpath = env.get("LD_LIBRARY_PATH", None)
self.repo.git.update_environment(LD_LIBRARY_PATH=libpath)
def close(self):
self.repo.close()
@property
def git(self):
return self.repo.git
def is_ignored(self, path: "Union[str, os.PathLike[str]]") -> bool:
from git.exc import GitCommandError
func = ignore(GitCommandError)(self.repo.git.check_ignore)
return bool(func(str(path)))
@property
def root_dir(self) -> str:
return self.repo.working_tree_dir
@staticmethod
def clone(
url: str,
to_path: str,
rev: Optional[str] = None,
shallow_branch: Optional[str] = None,
progress: Callable[["GitProgressEvent"], None] = None,
):
from git import Repo
from git.exc import GitCommandError
ld_key = "LD_LIBRARY_PATH"
env = fix_env()
if is_binary() and ld_key not in env.keys():
# In fix_env, we delete LD_LIBRARY_PATH key if it was empty before
# PyInstaller modified it. GitPython, in git.Repo.clone_from, uses
# env to update its own internal state. When there is no key in
# env, this value is not updated and GitPython re-uses
# LD_LIBRARY_PATH that has been set by PyInstaller.
# See [1] for more info.
# [1] https://github.com/gitpython-developers/GitPython/issues/924
env[ld_key] = ""
try:
if shallow_branch is not None and os.path.exists(url):
# git disables --depth for local clones unless file:// url
# scheme is used
url = f"file://{url}"
from dvc.scm.progress import GitProgressReporter
clone_from = partial(
Repo.clone_from,
url,
to_path,
env=env, # needed before we can fix it in __init__
no_single_branch=True,
progress=GitProgressReporter.wrap_fn(progress)
if progress
else None,
)
if shallow_branch is None:
tmp_repo = clone_from()
else:
tmp_repo = clone_from(branch=shallow_branch, depth=1)
tmp_repo.close()
except GitCommandError as exc: # pylint: disable=no-member
raise CloneError(url, to_path) from exc
# NOTE: using our wrapper to make sure that env is fixed in __init__
repo = GitPythonBackend(to_path)
if rev:
try:
repo.checkout(rev)
except GitCommandError as exc: # pylint: disable=no-member
raise RevError(
"failed to access revision '{}' for repo '{}'".format(
rev, url
)
) from exc
@staticmethod
def is_sha(rev):
import git
return rev and git.Repo.re_hexsha_shortened.search(rev)
@property
def dir(self) -> str:
return self.repo.git_dir
def add(self, paths: Union[str, Iterable[str]], update=False):
# NOTE: GitPython is not currently able to handle index version >= 3.
# See https://github.com/iterative/dvc/issues/610 for more details.
try:
if update:
if isinstance(paths, str):
paths = [paths]
self.git.add(*paths, update=True)
else:
self.repo.index.add(paths)
except AssertionError as exc:
raise UnsupportedIndexFormat from exc
def commit(self, msg: str, no_verify: bool = False):
from git.exc import HookExecutionError
try:
self.repo.index.commit(msg, skip_hooks=no_verify)
except HookExecutionError as exc:
raise SCMError("Git pre-commit hook failed") from exc
def checkout(
self,
branch: str,
create_new: Optional[bool] = False,
force: bool = False,
**kwargs,
):
if create_new:
self.repo.git.checkout("HEAD", b=branch, force=force, **kwargs)
else:
self.repo.git.checkout(branch, force=force, **kwargs)
def pull(self, **kwargs):
infos = self.repo.remote().pull(**kwargs)
for info in infos:
if info.flags & info.ERROR:
raise SCMError(f"pull failed: {info.note}")
def push(self):
infos = self.repo.remote().push()
for info in infos:
if info.flags & info.ERROR:
raise SCMError(f"push failed: {info.summary}")
def branch(self, branch):
self.repo.git.branch(branch)
def tag(self, tag):
self.repo.git.tag(tag)
def untracked_files(self):
files = self.repo.untracked_files
return [os.path.join(self.repo.working_dir, fname) for fname in files]
def is_tracked(self, path):
return bool(self.repo.git.ls_files(path))
def is_dirty(self, untracked_files: bool = False) -> bool:
return self.repo.is_dirty(untracked_files=untracked_files)
def active_branch(self):
return self.repo.active_branch.name
def list_branches(self):
return [h.name for h in self.repo.heads]
def list_tags(self):
return [t.name for t in self.repo.tags]
def list_all_commits(self):
head = self.get_ref("HEAD")
if not head:
# Empty repo
return []
return [
c.hexsha
for c in self.repo.iter_commits(
rev=head, branches=True, tags=True, remotes=True
)
]
def get_tree_obj(self, rev: str, **kwargs) -> GitPythonObject:
tree = self.repo.tree(rev)
return GitPythonObject(tree)
def get_rev(self):
return self.repo.rev_parse("HEAD").hexsha
def resolve_rev(self, rev):
from contextlib import suppress
from git.exc import BadName, GitCommandError
def _resolve_rev(name):
with suppress(BadName, GitCommandError):
try:
# Try python implementation of rev-parse first, it's faster
return self.repo.rev_parse(name).hexsha
except NotImplementedError:
# Fall back to `git rev-parse` for advanced features
return self.repo.git.rev_parse(name)
except ValueError:
raise RevError(f"unknown Git revision '{name}'")
# Resolve across local names
sha = _resolve_rev(rev)
if sha:
return sha
# Try all the remotes and if it resolves unambiguously then take it
if not self.is_sha(rev):
shas = {
_resolve_rev(f"{remote.name}/{rev}")
for remote in self.repo.remotes
} - {None}
if len(shas) > 1:
raise RevError(f"ambiguous Git revision '{rev}'")
if len(shas) == 1:
return shas.pop()
raise RevError(f"unknown Git revision '{rev}'")
def resolve_commit(self, rev: str) -> "GitCommit":
"""Return Commit object for the specified revision."""
from git.exc import BadName, GitCommandError
from git.objects.tag import TagObject
try:
commit = self.repo.rev_parse(rev)
except (BadName, GitCommandError):
raise SCMError(f"Invalid commit '{rev}'")
if isinstance(commit, TagObject):
commit = commit.object
return GitCommit(
commit.hexsha,
commit.committed_date,
commit.committer_tz_offset,
commit.message,
[str(parent) for parent in commit.parents],
)
def set_ref(
self,
name: str,
new_ref: str,
old_ref: Optional[str] = None,
message: Optional[str] = None,
symbolic: Optional[bool] = False,
):
from git.exc import GitCommandError
if old_ref and self.get_ref(name) != old_ref:
raise SCMError(f"Failed to set ref '{name}'")
try:
if symbolic:
if message:
self.git.symbolic_ref(name, new_ref, m=message)
else:
self.git.symbolic_ref(name, new_ref)
else:
args = [name, new_ref]
if old_ref:
args.append(old_ref)
if message:
self.git.update_ref(*args, m=message, create_reflog=True)
else:
self.git.update_ref(*args)
except GitCommandError as exc:
raise SCMError(f"Failed to set ref '{name}'") from exc
def get_ref(self, name: str, follow: bool = True) -> Optional[str]:
from git.exc import GitCommandError
if name == "HEAD":
try:
if follow or self.repo.head.is_detached:
return self.repo.head.commit.hexsha
return f"refs/heads/{self.repo.active_branch}"
except (GitCommandError, ValueError):
return None
elif name.startswith("refs/heads/"):
name = name[11:]
if name in self.repo.heads:
return self.repo.heads[name].commit.hexsha
elif name.startswith("refs/tags/"):
name = name[10:]
if name in self.repo.tags:
return self.repo.tags[name].commit.hexsha
else:
if not follow:
try:
rev = self.git.symbolic_ref(name).strip()
return rev if rev else None
except GitCommandError:
pass
try:
rev = self.git.show_ref(name, hash=True).strip()
return rev if rev else None
except GitCommandError:
pass
return None
def remove_ref(self, name: str, old_ref: Optional[str] = None):
from git.exc import GitCommandError
if old_ref and self.get_ref(name) != old_ref:
raise SCMError(f"Failed to set ref '{name}'")
try:
args = [name]
if old_ref:
args.append(old_ref)
self.git.update_ref(*args, d=True)
except GitCommandError as exc:
raise SCMError(f"Failed to set ref '{name}'") from exc
def iter_refs(self, base: Optional[str] = None):
from git import Reference
for ref in Reference.iter_items(self.repo, common_path=base):
yield ref.path
def iter_remote_refs(self, url: str, base: Optional[str] = None, **kwargs):
raise NotImplementedError
def get_refs_containing(self, rev: str, pattern: Optional[str] = None):
from git.exc import GitCommandError
try:
if pattern:
args = [pattern]
else:
args = []
for line in self.git.for_each_ref(
*args, contains=rev, format=r"%(refname)"
).splitlines():
line = line.strip()
if line:
yield line
except GitCommandError:
pass
def push_refspec(
self,
url: str,
src: Optional[str],
dest: str,
force: bool = False,
on_diverged: Optional[Callable[[str, str], bool]] = None,
progress: Callable[["GitProgressEvent"], None] = None,
**kwargs,
):
raise NotImplementedError
def fetch_refspecs(
self,
url: str,
refspecs: Iterable[str],
force: Optional[bool] = False,
on_diverged: Optional[Callable[[str, str], bool]] = None,
progress: Callable[["GitProgressEvent"], None] = None,
**kwargs,
):
raise NotImplementedError
def _stash_iter(self, ref: str):
raise NotImplementedError
def _stash_push(
self,
ref: str,
message: Optional[str] = None,
include_untracked: Optional[bool] = False,
) -> Tuple[Optional[str], bool]:
from dvc.scm.git import Stash
args = ["push"]
if message:
args.extend(["-m", message])
if include_untracked:
args.append("--include-untracked")
self.git.stash(*args)
commit = self.resolve_commit("stash@{0}")
if ref != Stash.DEFAULT_STASH:
# `git stash` CLI doesn't support using custom refspecs,
# so we push a commit onto refs/stash, make our refspec
# point to the new commit, then pop it from refs/stash
# `git stash create` is intended to be used for this kind of
# behavior but it doesn't support --include-untracked so we need to
# use push
self.set_ref(ref, commit.hexsha, message=commit.message)
self.git.stash("drop")
return commit.hexsha, False
def _stash_apply(self, rev: str):
from git.exc import GitCommandError
try:
self.git.stash("apply", rev)
except GitCommandError as exc:
out = str(exc)
if "CONFLICT" in out or "already exists" in out:
raise MergeConflictError(
"Stash apply resulted in merge conflicts"
) from exc
raise SCMError("Could not apply stash") from exc
def _stash_drop(self, ref: str, index: int):
from git.exc import GitCommandError
from dvc.scm.git import Stash
if ref == Stash.DEFAULT_STASH:
self.git.stash("drop", index)
return
self.git.reflog(
"delete", "--updateref", "--rewrite", f"{ref}@{{{index}}}"
)
try:
self.git.reflog("exists", ref)
except GitCommandError:
self.remove_ref(ref)
def describe(
self,
rev: str,
base: Optional[str] = None,
match: Optional[str] = None,
exclude: Optional[str] = None,
) -> Optional[str]:
raise NotImplementedError
def diff(self, rev_a: str, rev_b: str, binary=False) -> str:
raise NotImplementedError
def reset(self, hard: bool = False, paths: Iterable[str] = None):
if paths:
paths_list: Optional[List[str]] = [
relpath(path, self.root_dir) for path in paths
]
if os.name == "nt":
paths_list = [
path.replace("\\", "/")
for path in paths_list # type: ignore[union-attr]
]
else:
paths_list = None
self.repo.head.reset(index=True, working_tree=hard, paths=paths_list)
def checkout_index(
self,
paths: Optional[Iterable[str]] = None,
force: bool = False,
ours: bool = False,
theirs: bool = False,
):
"""Checkout the specified paths from HEAD index."""
assert not (ours and theirs)
if ours or theirs:
args = ["--ours"] if ours else ["--theirs"]
if force:
args.append("--force")
args.append("--")
if paths:
args.extend(list(paths))
else:
args.append(".")
self.repo.git.checkout(*args)
else:
if paths:
paths_list: Optional[List[str]] = [
relpath(path, self.root_dir) for path in paths
]
if os.name == "nt":
paths_list = [
path.replace("\\", "/")
for path in paths_list # type: ignore[union-attr]
]
else:
paths_list = None
self.repo.index.checkout(paths=paths_list, force=force)
def status(
self, ignored: bool = False
) -> Tuple[Mapping[str, Iterable[str]], Iterable[str], Iterable[str]]:
raise NotImplementedError
def merge(
self,
rev: str,
commit: bool = True,
msg: Optional[str] = None,
squash: bool = False,
) -> Optional[str]:
from git.exc import GitCommandError
if commit and squash:
raise SCMError("Cannot merge with 'squash' and 'commit'")
if commit and not msg:
raise SCMError("Merge commit message is required")
merge = partial(self.git.merge, rev)
try:
if commit:
merge(m=msg)
return self.get_rev()
merge(no_commit=True, squash=True)
except GitCommandError as exc:
if "CONFLICT" in str(exc):
raise MergeConflictError("Merge contained conflicts") from exc
raise SCMError("Merge failed") from exc
return None
def validate_git_remote(self, url: str, **kwargs):
raise NotImplementedError
-592
View File
@@ -1,592 +0,0 @@
import locale
import logging
import os
import stat
from contextlib import contextmanager
from io import BytesIO, StringIO
from typing import (
TYPE_CHECKING,
Callable,
Iterable,
List,
Mapping,
Optional,
Tuple,
Union,
)
from funcy import cached_property
from dvc.scm.exceptions import MergeConflictError, RevError, SCMError
from dvc.scm.utils import relpath
from ..objects import GitCommit, GitObject
from .base import BaseGitBackend
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from dvc.scm.progress import GitProgressEvent
# NOTE: constant from libgit2 git2/checkout.h
# This can be removed after next pygit2 release:
# see https://github.com/libgit2/pygit2/pull/1087
GIT_CHECKOUT_SKIP_LOCKED_DIRECTORIES = 1 << 18
class Pygit2Object(GitObject):
def __init__(self, obj):
self.obj = obj
def open(self, mode: str = "r", encoding: str = None):
if not encoding:
encoding = locale.getpreferredencoding(False)
data = self.obj.read_raw()
if mode == "rb":
return BytesIO(data)
return StringIO(data.decode(encoding))
@property
def name(self) -> str:
return self.obj.name
@property
def mode(self):
if not self.obj.filemode and self.obj.type_str == "tree":
return stat.S_IFDIR
return self.obj.filemode
@property
def size(self) -> int:
return len(self.obj.read_raw())
@property
def sha(self) -> str:
return self.obj.hex
def scandir(self) -> Iterable["Pygit2Object"]:
for entry in self.obj: # noqa: B301
yield Pygit2Object(entry)
class Pygit2Backend(BaseGitBackend): # pylint:disable=abstract-method
def __init__( # pylint:disable=W0231
self, root_dir=os.curdir, search_parent_directories=True
):
import pygit2
if search_parent_directories:
ceiling_dirs = ""
else:
ceiling_dirs = os.path.abspath(root_dir)
# NOTE: discover_repository will return path/.git/
path = pygit2.discover_repository( # pylint:disable=no-member
root_dir, True, ceiling_dirs
)
if not path:
raise SCMError(f"{root_dir} is not a git repository")
self.repo = pygit2.Repository(path)
self._stashes: dict = {}
def close(self):
if hasattr(self, "_refdb"):
del self._refdb
self.repo.free()
@property
def root_dir(self) -> str:
return self.repo.workdir
@cached_property
def _refdb(self):
from pygit2 import RefdbFsBackend
return RefdbFsBackend(self.repo)
def _resolve_refish(self, refish: str):
from pygit2 import GIT_OBJ_COMMIT, Tag
commit, ref = self.repo.resolve_refish(refish)
if isinstance(commit, Tag):
ref = commit
commit = commit.peel(GIT_OBJ_COMMIT)
return commit, ref
@property
def default_signature(self):
try:
return self.repo.default_signature
except KeyError as exc:
raise SCMError(
"Git username and email must be configured"
) from exc
@staticmethod
def _get_checkout_strategy(strategy: Optional[int] = None):
from pygit2 import GIT_CHECKOUT_RECREATE_MISSING, GIT_CHECKOUT_SAFE
if strategy is None:
strategy = GIT_CHECKOUT_SAFE | GIT_CHECKOUT_RECREATE_MISSING
if os.name == "nt":
strategy |= GIT_CHECKOUT_SKIP_LOCKED_DIRECTORIES
return strategy
# Workaround to force git_backend_odb_pack to release open file handles
# in DVC's mixed git-backend environment.
# See https://github.com/iterative/dvc/issues/5641
@contextmanager
def release_odb_handles(self):
yield
# It is safe to free the libgit2 repo context multiple times - free
# just forces libgit/pygit to release git ODB related contexts which
# can be reacquired later as needed.
self.repo.free()
@staticmethod
def clone(
url: str,
to_path: str,
rev: Optional[str] = None,
shallow_branch: Optional[str] = None,
progress: Callable[["GitProgressEvent"], None] = None,
):
raise NotImplementedError
@property
def dir(self) -> str:
raise NotImplementedError
def add(self, paths: Union[str, Iterable[str]], update=False):
raise NotImplementedError
def commit(self, msg: str, no_verify: bool = False):
raise NotImplementedError
def checkout(
self,
branch: str,
create_new: Optional[bool] = False,
force: bool = False,
**kwargs,
):
from pygit2 import GIT_CHECKOUT_FORCE, GitError
strategy = self._get_checkout_strategy(
GIT_CHECKOUT_FORCE if force else None
)
with self.release_odb_handles():
if create_new:
commit = self.repo.revparse_single("HEAD")
new_branch = self.repo.branches.local.create(branch, commit)
self.repo.checkout(new_branch, strategy=strategy)
else:
if branch == "-":
branch = "@{-1}"
try:
commit, ref = self._resolve_refish(branch)
except (KeyError, GitError):
raise RevError(f"unknown Git revision '{branch}'")
self.repo.checkout_tree(commit, strategy=strategy)
detach = kwargs.get("detach", False)
if ref and not detach:
self.repo.set_head(ref.name)
else:
self.repo.set_head(commit.id)
def pull(self, **kwargs):
raise NotImplementedError
def push(self):
raise NotImplementedError
def branch(self, branch: str):
from pygit2 import GitError
try:
commit = self.repo[self.repo.head.target]
self.repo.create_branch(branch, commit)
except GitError as exc:
raise SCMError(f"Failed to create branch '{branch}'") from exc
def tag(self, tag: str):
raise NotImplementedError
def untracked_files(self) -> Iterable[str]:
raise NotImplementedError
def is_tracked(self, path: str) -> bool:
raise NotImplementedError
def is_dirty(self, untracked_files: bool = False) -> bool:
raise NotImplementedError
def active_branch(self) -> str:
raise NotImplementedError
def list_branches(self) -> Iterable[str]:
raise NotImplementedError
def list_tags(self) -> Iterable[str]:
raise NotImplementedError
def list_all_commits(self) -> Iterable[str]:
raise NotImplementedError
def get_tree_obj(self, rev: str, **kwargs) -> Pygit2Object:
tree = self.repo[rev].tree
return Pygit2Object(tree)
def get_rev(self) -> str:
raise NotImplementedError
def resolve_rev(self, rev: str) -> str:
from pygit2 import GitError
try:
commit, _ref = self._resolve_refish(rev)
return str(commit.id)
except (KeyError, GitError):
pass
# Look for single exact match in remote refs
shas = {
self.get_ref(f"refs/remotes/{remote.name}/{rev}")
for remote in self.repo.remotes
} - {None}
if len(shas) > 1:
raise RevError(f"ambiguous Git revision '{rev}'")
if len(shas) == 1:
return shas.pop() # type: ignore
raise RevError(f"unknown Git revision '{rev}'")
def resolve_commit(self, rev: str) -> "GitCommit":
from pygit2 import GitError
try:
commit, _ref = self._resolve_refish(rev)
except (KeyError, GitError):
raise SCMError(f"Invalid commit '{rev}'")
return GitCommit(
str(commit.id),
commit.commit_time,
commit.commit_time_offset,
commit.message,
[str(parent) for parent in commit.parent_ids],
)
def _get_stash(self, ref: str):
raise NotImplementedError
def is_ignored(self, path: "Union[str, os.PathLike[str]]") -> bool:
rel = relpath(path, self.root_dir)
if os.name == "nt":
rel.replace("\\", "/")
return self.repo.path_is_ignored(rel)
def set_ref(
self,
name: str,
new_ref: str,
old_ref: Optional[str] = None,
message: Optional[str] = None,
symbolic: Optional[bool] = False,
):
if old_ref and old_ref != self.get_ref(name, follow=False):
raise SCMError(f"Failed to set '{name}'")
if message:
self._refdb.ensure_log(name)
if symbolic:
self.repo.create_reference_symbolic(
name, new_ref, True, message=message
)
else:
self.repo.create_reference_direct(
name, new_ref, True, message=message
)
def get_ref(self, name, follow: bool = True) -> Optional[str]:
from pygit2 import GIT_REF_SYMBOLIC
ref = self.repo.references.get(name)
if not ref:
return None
if follow and ref.type == GIT_REF_SYMBOLIC:
ref = ref.resolve()
return str(ref.target)
def remove_ref(self, name: str, old_ref: Optional[str] = None):
ref = self.repo.references.get(name)
if not ref:
raise SCMError(f"Ref '{name}' does not exist")
if old_ref and old_ref != str(ref.target):
raise SCMError(f"Failed to remove '{name}'")
ref.delete()
def iter_refs(self, base: Optional[str] = None):
if base:
for ref in self.repo.references:
if ref.startswith(base):
yield ref
else:
yield from self.repo.references
def get_refs_containing(self, rev: str, pattern: Optional[str] = None):
import fnmatch
from pygit2 import GitError
def _contains(repo, ref, search_commit):
commit, _ref = self._resolve_refish(ref)
base = repo.merge_base(search_commit.id, commit.id)
return base == search_commit.id
try:
search_commit, _ref = self._resolve_refish(rev)
except (KeyError, GitError):
raise SCMError(f"Invalid rev '{rev}'")
if not pattern:
yield from (
ref
for ref in self.iter_refs()
if _contains(self.repo, ref, search_commit)
)
return
literal = pattern.rstrip("/").split("/")
for ref in self.iter_refs():
if (
ref.split("/")[: len(literal)] == literal
or fnmatch.fnmatch(ref, pattern)
) and _contains(self.repo, ref, search_commit):
yield ref
def push_refspec(
self,
url: str,
src: Optional[str],
dest: str,
force: bool = False,
on_diverged: Optional[Callable[[str, str], bool]] = None,
progress: Callable[["GitProgressEvent"], None] = None,
**kwargs,
):
raise NotImplementedError
def fetch_refspecs(
self,
url: str,
refspecs: Iterable[str],
force: Optional[bool] = False,
on_diverged: Optional[Callable[[str, str], bool]] = None,
progress: Callable[["GitProgressEvent"], None] = None,
**kwargs,
):
raise NotImplementedError
def _stash_iter(self, ref: str):
raise NotImplementedError
def _stash_push(
self,
ref: str,
message: Optional[str] = None,
include_untracked: Optional[bool] = False,
) -> Tuple[Optional[str], bool]:
from dvc.scm.git import Stash
oid = self.repo.stash(
self.default_signature,
message=message,
include_untracked=include_untracked,
)
commit = self.repo[oid]
if ref != Stash.DEFAULT_STASH:
self.set_ref(ref, commit.id, message=commit.message)
self.repo.stash_drop()
return str(oid), False
def _stash_apply(self, rev: str):
from pygit2 import GitError
from dvc.scm.git import Stash
def _apply(index):
try:
self.repo.index.read(False)
self.repo.stash_apply(
index, strategy=self._get_checkout_strategy()
)
except GitError as exc:
raise MergeConflictError(
"Stash apply resulted in merge conflicts"
) from exc
# libgit2 stash apply only accepts refs/stash items by index. If rev is
# not in refs/stash, we will push it onto the stash, and then pop it
commit, _ref = self._resolve_refish(rev)
stash = self.repo.references.get(Stash.DEFAULT_STASH)
if stash:
for i, entry in enumerate(stash.log()):
if entry.oid_new == commit.id:
_apply(i)
return
self.set_ref(Stash.DEFAULT_STASH, commit.id, message=commit.message)
try:
_apply(0)
finally:
self.repo.stash_drop()
def _stash_drop(self, ref: str, index: int):
from dvc.scm.git import Stash
if ref != Stash.DEFAULT_STASH:
raise NotImplementedError
self.repo.stash_drop(index)
def describe(
self,
rev: str,
base: Optional[str] = None,
match: Optional[str] = None,
exclude: Optional[str] = None,
) -> Optional[str]:
raise NotImplementedError
def diff(self, rev_a: str, rev_b: str, binary=False) -> str:
raise NotImplementedError
def reset(self, hard: bool = False, paths: Iterable[str] = None):
from pygit2 import GIT_RESET_HARD, GIT_RESET_MIXED, IndexEntry
self.repo.index.read(False)
if paths is not None:
tree = self.repo.revparse_single("HEAD").tree
for path in paths:
rel = relpath(path, self.root_dir)
if os.name == "nt":
rel = rel.replace("\\", "/")
obj = tree[rel]
self.repo.index.add(IndexEntry(rel, obj.oid, obj.filemode))
self.repo.index.write()
elif hard:
self.repo.reset(self.repo.head.target, GIT_RESET_HARD)
else:
self.repo.reset(self.repo.head.target, GIT_RESET_MIXED)
def checkout_index(
self,
paths: Optional[Iterable[str]] = None,
force: bool = False,
ours: bool = False,
theirs: bool = False,
):
from pygit2 import (
GIT_CHECKOUT_ALLOW_CONFLICTS,
GIT_CHECKOUT_FORCE,
GIT_CHECKOUT_RECREATE_MISSING,
GIT_CHECKOUT_SAFE,
)
assert not (ours and theirs)
strategy = GIT_CHECKOUT_RECREATE_MISSING
if force or ours or theirs:
strategy |= GIT_CHECKOUT_FORCE
else:
strategy |= GIT_CHECKOUT_SAFE
if ours or theirs:
strategy |= GIT_CHECKOUT_ALLOW_CONFLICTS
strategy = self._get_checkout_strategy(strategy)
index = self.repo.index
if paths:
path_list: Optional[List[str]] = [
relpath(path, self.root_dir) for path in paths
]
if os.name == "nt":
path_list = [
path.replace("\\", "/")
for path in path_list # type: ignore[union-attr]
]
else:
path_list = None
with self.release_odb_handles():
self.repo.checkout_index(
index=index, paths=path_list, strategy=strategy
)
if index.conflicts and (ours or theirs):
for ancestor, ours_entry, theirs_entry in index.conflicts:
if not ancestor:
continue
if ours:
entry = ours_entry
index.add(ours_entry)
else:
entry = theirs_entry
path = os.path.join(self.root_dir, entry.path)
with open(path, "wb") as fobj:
fobj.write(self.repo.get(entry.id).read_raw())
index.add(entry.path)
index.write()
def iter_remote_refs(self, url: str, base: Optional[str] = None, **kwargs):
raise NotImplementedError
def status(
self, ignored: bool = False
) -> Tuple[Mapping[str, Iterable[str]], Iterable[str], Iterable[str]]:
raise NotImplementedError
def merge(
self,
rev: str,
commit: bool = True,
msg: Optional[str] = None,
squash: bool = False,
) -> Optional[str]:
from pygit2 import GIT_RESET_MIXED, GitError
if commit and squash:
raise SCMError("Cannot merge with 'squash' and 'commit'")
if commit and not msg:
raise SCMError("Merge commit message is required")
with self.release_odb_handles():
try:
self.repo.index.read(False)
self.repo.merge(rev)
self.repo.index.write()
except GitError as exc:
raise SCMError("Merge failed") from exc
if self.repo.index.conflicts:
raise MergeConflictError("Merge contained conflicts")
if commit:
user = self.default_signature
tree = self.repo.index.write_tree()
merge_commit = self.repo.create_commit(
"HEAD", user, user, msg, tree, [self.repo.head.target, rev]
)
return str(merge_commit)
if squash:
self.repo.reset(self.repo.head.target, GIT_RESET_MIXED)
self.repo.state_cleanup()
self.repo.index.write()
return None
def validate_git_remote(self, url: str, **kwargs):
raise NotImplementedError
-144
View File
@@ -1,144 +0,0 @@
import stat
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Iterable, List, Optional
from pygtrie import Trie
S_IFGITLINK = 0o160000
def S_ISGITLINK(m: int) -> bool:
return stat.S_IFMT(m) == S_IFGITLINK
class GitObject(ABC):
@abstractmethod
def open(self, mode: str = "r", encoding: str = None):
pass
@property
@abstractmethod
def name(self) -> str:
pass
@property
@abstractmethod
def mode(self) -> int:
pass
@abstractmethod
def scandir(self) -> Iterable["GitObject"]:
pass
@property
def isfile(self) -> bool:
return stat.S_ISREG(self.mode)
@property
def isdir(self) -> bool:
return stat.S_ISDIR(self.mode)
@property
def issubmodule(self) -> bool:
return S_ISGITLINK(self.mode)
@property
@abstractmethod
def size(self) -> int:
pass
@property
@abstractmethod
def sha(self) -> str:
pass
class GitTrie:
def __init__(self, tree: GitObject, rev: str):
self.tree = tree
self.rev = rev
self.trie = Trie()
self.trie[()] = tree
self._build(tree, ())
def _build(self, tree: GitObject, path: tuple):
for obj in tree.scandir():
obj_path = path + (obj.name,)
self.trie[obj_path] = obj
if obj.isdir:
self._build(obj, obj_path)
def open(
self,
key: tuple,
mode: Optional[str] = "r",
encoding: Optional[str] = None,
):
obj = self.trie[key]
if obj.isdir:
raise IsADirectoryError
return obj.open(mode=mode, encoding=encoding)
def exists(self, key: tuple) -> bool:
return bool(self.trie.has_node(key))
def isdir(self, key: tuple) -> bool:
try:
obj = self.trie[key]
except KeyError:
return False
return obj.isdir
def isfile(self, key: tuple) -> bool:
try:
obj = self.trie[key]
except KeyError:
return False
return obj.isfile
def walk(self, top: tuple, topdown: Optional[bool] = True):
dirs = []
nondirs = []
def node_factory(_, path, children, obj):
if path == top:
assert obj.isdir
list(filter(None, children))
elif obj.isdir:
dirs.append(obj.name)
else:
nondirs.append(obj.name)
self.trie.traverse(node_factory, prefix=top)
if topdown:
yield top, dirs, nondirs
for dname in dirs:
yield from self.walk(top + (dname,), topdown=topdown)
if not topdown:
yield top, dirs, nondirs
def info(self, key: tuple) -> dict:
obj = self.trie[key]
return {
"size": obj.size,
"type": "directory" if stat.S_ISDIR(obj.mode) else "file",
"sha": obj.sha,
"mode": obj.mode,
}
@dataclass
class GitCommit:
hexsha: str
commit_time: int
commit_time_offset: int
message: str
parents: List[str]
-73
View File
@@ -1,73 +0,0 @@
"""git stash convenience wrapper."""
import logging
from typing import Optional
from dvc.scm.exceptions import SCMError
logger = logging.getLogger(__name__)
class Stash:
"""Wrapper for representing any arbitrary Git stash ref."""
DEFAULT_STASH = "refs/stash"
def __init__(self, scm, ref: Optional[str] = None):
self.ref = ref if ref else self.DEFAULT_STASH
self.scm = scm
def __iter__(self):
yield from self.scm._stash_iter(self.ref)
def __len__(self):
return len(self.list())
def __getitem__(self, index):
return self.list()[index]
def list(self):
return list(iter(self))
def push(
self, message: Optional[str] = None, include_untracked: bool = False
) -> Optional[str]:
if not self.scm.is_dirty(untracked_files=include_untracked):
logger.debug("No changes to stash")
return None
logger.debug("Stashing changes in '%s'", self.ref)
rev, reset = self.scm._stash_push( # pylint: disable=protected-access
self.ref, message=message, include_untracked=include_untracked
)
if reset:
self.scm.reset(hard=True)
return rev
def pop(self):
logger.debug("Popping from stash '%s'", self.ref)
ref = f"{self.ref}@{{0}}"
rev = self.scm.resolve_rev(ref)
try:
self.apply(rev)
except Exception as exc:
raise SCMError("Could not apply stash commit") from exc
self.drop()
return rev
def apply(self, rev):
logger.debug("Applying stash commit '%s'", rev)
self.scm._stash_apply(rev) # pylint: disable=protected-access
def drop(self, index: int = 0):
if index < 0 or index >= len(self):
raise SCMError(f"Invalid stash ref '{self.ref}@{{{index}}}'")
logger.debug("Dropping '%s@{%d}'", self.ref, index)
self.scm._stash_drop( # pylint: disable=protected-access
self.ref, index
)
def clear(self):
logger.debug("Clear stash '%s'", self.ref)
for _ in range(len(self)):
self.drop()
-16
View File
@@ -1,16 +0,0 @@
from .base import Base
# Syntactic sugar to signal that this is an actual implementation for a DVC
# project under no SCM control.
class NoSCM(Base):
def __init__(
self,
root_dir: str = None,
_raise_not_implemented_as=NotImplementedError,
):
super().__init__(root_dir=root_dir)
self._exc = _raise_not_implemented_as
def __getattr__(self, name):
raise self._exc
-73
View File
@@ -1,73 +0,0 @@
from typing import NamedTuple, Optional, Union
from funcy import compose
def code2desc(op_code):
from git import RootUpdateProgress as OP
ops = {
OP.COUNTING: "Counting",
OP.COMPRESSING: "Compressing",
OP.WRITING: "Writing",
OP.RECEIVING: "Receiving",
OP.RESOLVING: "Resolving",
OP.FINDING_SOURCES: "Finding sources",
OP.CHECKING_OUT: "Checking out",
OP.CLONE: "Cloning",
OP.FETCH: "Fetching",
OP.UPDWKTREE: "Updating working tree",
OP.REMOVE: "Removing",
OP.PATHCHANGE: "Changing path",
OP.URLCHANGE: "Changing URL",
OP.BRANCHCHANGE: "Changing branch",
}
return ops.get(op_code & OP.OP_MASK, "")
class GitProgressEvent(NamedTuple):
phase: str = ""
completed: Optional[int] = None
total: Optional[int] = None
message: str = ""
@classmethod
def parsed_from_gitpython(
cls,
op_code,
cur_count,
max_count=None,
message="", # pylint: disable=redefined-outer-name
):
return cls(code2desc(op_code), cur_count, max_count, message)
class GitProgressReporter:
def __init__(self, fn) -> None:
from git.util import CallableRemoteProgress
self._reporter = CallableRemoteProgress(self.wrap_fn(fn))
def __call__(self, msg: Union[str, bytes]) -> None:
self._reporter._parse_progress_line(
msg.decode("ascii").strip() if isinstance(msg, bytes) else msg
)
@staticmethod
def wrap_fn(fn):
return compose(fn, GitProgressEvent.parsed_from_gitpython)
if __name__ == "__main__":
message = """
Cloning into 'dvcyaml-schema'...
remote: Enumerating objects: 76, done.
remote: Counting objects: 100% (76/76), done.
remote: Compressing objects: 100% (56/56), done.
remote: Total 76 (delta 30), reused 38 (delta 9), pack-reused 0
Receiving objects: 100% (76/76), 24.83 KiB | 89.00 KiB/s, done.
Resolving deltas: 100% (30/30), done."""
reporter = GitProgressReporter(print)
for line in message.splitlines():
reporter(line)
-22
View File
@@ -1,22 +0,0 @@
import os
def relpath(path, start=os.curdir):
path = os.fspath(path)
start = os.path.abspath(os.fspath(start))
# Windows path on different drive than curdir doesn't have relpath
if os.name == "nt":
# Since python 3.8 os.realpath resolves network shares to their UNC
# path. So, to be certain that relative paths correctly captured,
# we need to resolve to UNC path first. We resolve only the drive
# name so that we don't follow any 'real' symlinks on the path
def resolve_network_drive_windows(path_to_resolve):
drive, tail = os.path.splitdrive(path_to_resolve)
return os.path.join(os.path.realpath(drive), tail)
path = resolve_network_drive_windows(os.path.abspath(path))
start = resolve_network_drive_windows(start)
if not os.path.commonprefix([start, path]):
return path
return os.path.relpath(path, start)
+1 -3
View File
@@ -37,9 +37,6 @@ install_requires =
ply>=3.9
colorama>=0.3.9
configobj>=5.0.6
gitpython>3
dulwich>=0.20.23
pygit2>=1.5.0
nanotime>=0.5.2
pyasn1>=0.4.1
voluptuous>=0.11.7
@@ -76,6 +73,7 @@ install_requires =
aiohttp-retry>=2.4.5
diskcache>=5.2.1
jaraco.windows>=5.7.0; python_version < '3.8' and sys_platform == 'win32'
scmrepo==0.0.1
[options.extras_require]
# gssapi should not be included in all_remotes, because it doesn't have wheels
+4 -2
View File
@@ -104,8 +104,9 @@ class TmpDir(pathlib.Path):
return self
def init(self, *, scm=False, dvc=False, subdir=False):
from scmrepo.git import Git
from dvc.repo import Repo
from dvc.scm.git import Git
assert not scm or not hasattr(self, "scm")
assert not dvc or not hasattr(self, "dvc")
@@ -304,8 +305,9 @@ def make_tmp_dir(tmp_path_factory, request, worker_id):
def make(name, *, scm=False, dvc=False, subdir=False):
from shutil import ignore_patterns
from scmrepo.git import Git
from dvc.repo import Repo
from dvc.scm.git import Git
cache = CACHE.get((scm, dvc, subdir))
if not cache:
+2 -1
View File
@@ -145,8 +145,9 @@ def test_metrics_diff_with_unchanged(tmp_dir, scm, dvc, run_copy_metrics):
def test_no_commits(tmp_dir):
from scmrepo.git import Git
from dvc.repo import Repo
from dvc.scm.git import Git
from tests.dir_helpers import git_init
git_init(".")
+2 -1
View File
@@ -126,8 +126,9 @@ def test_pipeline_tracked_params(tmp_dir, scm, dvc, run_copy):
def test_no_commits(tmp_dir):
from scmrepo.git import Git
from dvc.repo import Repo
from dvc.scm.git import Git
from tests.dir_helpers import git_init
git_init(".")
+2 -1
View File
@@ -268,8 +268,9 @@ def test_no_changes(tmp_dir, scm, dvc):
def test_no_commits(tmp_dir):
from scmrepo.git import Git
from dvc.repo import Repo
from dvc.scm.git import Git
from tests.dir_helpers import git_init
git_init(".")
+2 -1
View File
@@ -1,10 +1,11 @@
import os
from unittest.mock import ANY, patch
from scmrepo.git import Git
from dvc.external_repo import CLONES, external_repo
from dvc.objects.stage import stage
from dvc.objects.transfer import transfer
from dvc.scm.git import Git
from dvc.utils import relpath
from dvc.utils.fs import makedirs, remove
from tests.unit.fs.test_repo import make_subrepo
+1 -1
View File
@@ -474,7 +474,7 @@ def test_download_dir_callback(tmp_dir, dvc, cloud):
@pytest.mark.parametrize("fs_type", ["git", "dvc"])
def test_download_callbacks_on_dvc_git_fs(tmp_dir, dvc, scm, fs_type):
from dvc.fs.scm import GitFileSystem
from dvc.fs.git import GitFileSystem
gen = tmp_dir.scm_gen if fs_type == "git" else tmp_dir.dvc_gen
gen({"dir": {"foo": "foo", "bar": "bar"}, "file": "file"}, commit="gen")
+1 -1
View File
@@ -123,7 +123,7 @@ def test_ignore_collecting_dvcignores(tmp_dir, dvc, dname):
def test_ignore_on_branch(tmp_dir, scm, dvc):
from dvc.fs.scm import GitFileSystem
from dvc.fs.git import GitFileSystem
tmp_dir.scm_gen({"foo": "foo", "bar": "bar"}, commit="add files")
+4 -4
View File
@@ -230,7 +230,7 @@ def test_git_stash_workspace(tmp_dir, scm):
],
)
def test_git_stash_push(tmp_dir, scm, ref, include_untracked):
from dvc.scm.git import Stash
from scmrepo.git import Stash
tmp_dir.scm_gen({"file": "0"}, commit="init")
tmp_dir.gen({"file": "1", "untracked": "0"})
@@ -255,7 +255,7 @@ def test_git_stash_push(tmp_dir, scm, ref, include_untracked):
@pytest.mark.parametrize("ref", [None, "refs/foo/stash"])
def test_git_stash_drop(tmp_dir, scm, ref):
from dvc.scm.git import Stash
from scmrepo.git import Stash
tmp_dir.scm_gen({"file": "0"}, commit="init")
tmp_dir.gen("file", "1")
@@ -288,7 +288,7 @@ reason = """libgit2 stash_save() is flaky on linux when run inside pytest
],
)
def test_git_stash_pop(tmp_dir, scm, ref):
from dvc.scm.git import Stash
from scmrepo.git import Stash
tmp_dir.scm_gen({"file": "0"}, commit="init")
tmp_dir.gen("file", "1")
@@ -307,7 +307,7 @@ def test_git_stash_pop(tmp_dir, scm, ref):
@pytest.mark.parametrize("ref", [None, "refs/foo/stash"])
def test_git_stash_clear(tmp_dir, scm, ref):
from dvc.scm.git import Stash
from scmrepo.git import Stash
tmp_dir.scm_gen({"file": "0"}, commit="init")
tmp_dir.gen("file", "1")
+8 -9
View File
@@ -1,15 +1,14 @@
import os
import pytest
from dvc.scm.exceptions import SCMError
from scmrepo.exceptions import SCMError
# Behaves the same as SCM but will test against all supported Git backends.
# tmp_dir.scm will still contain a default SCM instance.
@pytest.fixture(params=["gitpython", "dulwich", "pygit2"])
def git(tmp_dir, scm, request):
from dvc.scm.git import Git
from scmrepo.git import Git
git_ = Git(os.fspath(tmp_dir), backends=[request.param])
git_.test_backend = request.param
@@ -109,7 +108,8 @@ def test_is_tracked_unicode(tmp_dir, scm):
def test_no_commits(tmp_dir):
from dvc.scm.git import Git
from scmrepo.git import Git
from tests.dir_helpers import git_init
git_init(".")
@@ -344,7 +344,7 @@ def test_commit_no_verify(tmp_dir, scm, git, hook):
@pytest.mark.parametrize("squash", [True, False])
def test_merge(tmp_dir, scm, git, squash):
from dvc.scm.exceptions import MergeConflictError
from scmrepo.exceptions import MergeConflictError
if git.test_backend == "dulwich":
pytest.skip()
@@ -399,7 +399,7 @@ def test_checkout_index(tmp_dir, scm, git):
"strategy, expected", [("ours", "baz"), ("theirs", "bar")]
)
def test_checkout_index_conflicts(tmp_dir, scm, git, strategy, expected):
from dvc.scm.exceptions import MergeConflictError
from scmrepo.exceptions import MergeConflictError
if git.test_backend == "dulwich":
pytest.skip()
@@ -422,7 +422,7 @@ def test_checkout_index_conflicts(tmp_dir, scm, git, strategy, expected):
def test_resolve_rev(tmp_dir, scm, make_tmp_dir, git):
from dvc.scm.exceptions import RevError
from scmrepo.exceptions import RevError
if git.test_backend == "dulwich":
pytest.skip()
@@ -595,8 +595,7 @@ def test_pygit_checkout_subdir(tmp_dir, scm, git):
)
def test_dulwich_github_compat(mocker, algorithm):
from asyncssh.misc import ProtocolError
from dvc.scm.git.backend.dulwich.asyncssh_vendor import (
from scmrepo.git.backend.dulwich.asyncssh_vendor import (
_process_public_key_ok_gh,
)