Spaces:
Sleeping
Sleeping
audio-transcription-app
/
venv
/lib
/python3.11
/site-packages
/pip
/_internal
/vcs
/versioncontrol.py
"""Handles all VCS (version control) support""" | |
import logging | |
import os | |
import shutil | |
import sys | |
import urllib.parse | |
from typing import ( | |
TYPE_CHECKING, | |
Any, | |
Dict, | |
Iterable, | |
Iterator, | |
List, | |
Mapping, | |
Optional, | |
Tuple, | |
Type, | |
Union, | |
) | |
from pip._internal.cli.spinners import SpinnerInterface | |
from pip._internal.exceptions import BadCommand, InstallationError | |
from pip._internal.utils.misc import ( | |
HiddenText, | |
ask_path_exists, | |
backup_dir, | |
display_path, | |
hide_url, | |
hide_value, | |
is_installable_dir, | |
rmtree, | |
) | |
from pip._internal.utils.subprocess import ( | |
CommandArgs, | |
call_subprocess, | |
format_command_args, | |
make_command, | |
) | |
from pip._internal.utils.urls import get_url_scheme | |
if TYPE_CHECKING: | |
# Literal was introduced in Python 3.8. | |
# | |
# TODO: Remove `if TYPE_CHECKING` when dropping support for Python 3.7. | |
from typing import Literal | |
__all__ = ["vcs"] | |
logger = logging.getLogger(__name__) | |
AuthInfo = Tuple[Optional[str], Optional[str]] | |
def is_url(name: str) -> bool: | |
""" | |
Return true if the name looks like a URL. | |
""" | |
scheme = get_url_scheme(name) | |
if scheme is None: | |
return False | |
return scheme in ["http", "https", "file", "ftp"] + vcs.all_schemes | |
def make_vcs_requirement_url( | |
repo_url: str, rev: str, project_name: str, subdir: Optional[str] = None | |
) -> str: | |
""" | |
Return the URL for a VCS requirement. | |
Args: | |
repo_url: the remote VCS url, with any needed VCS prefix (e.g. "git+"). | |
project_name: the (unescaped) project name. | |
""" | |
egg_project_name = project_name.replace("-", "_") | |
req = f"{repo_url}@{rev}#egg={egg_project_name}" | |
if subdir: | |
req += f"&subdirectory={subdir}" | |
return req | |
def find_path_to_project_root_from_repo_root( | |
location: str, repo_root: str | |
) -> Optional[str]: | |
""" | |
Find the the Python project's root by searching up the filesystem from | |
`location`. Return the path to project root relative to `repo_root`. | |
Return None if the project root is `repo_root`, or cannot be found. | |
""" | |
# find project root. | |
orig_location = location | |
while not is_installable_dir(location): | |
last_location = location | |
location = os.path.dirname(location) | |
if location == last_location: | |
# We've traversed up to the root of the filesystem without | |
# finding a Python project. | |
logger.warning( | |
"Could not find a Python project for directory %s (tried all " | |
"parent directories)", | |
orig_location, | |
) | |
return None | |
if os.path.samefile(repo_root, location): | |
return None | |
return os.path.relpath(location, repo_root) | |
class RemoteNotFoundError(Exception): | |
pass | |
class RemoteNotValidError(Exception): | |
def __init__(self, url: str): | |
super().__init__(url) | |
self.url = url | |
class RevOptions: | |
""" | |
Encapsulates a VCS-specific revision to install, along with any VCS | |
install options. | |
Instances of this class should be treated as if immutable. | |
""" | |
def __init__( | |
self, | |
vc_class: Type["VersionControl"], | |
rev: Optional[str] = None, | |
extra_args: Optional[CommandArgs] = None, | |
) -> None: | |
""" | |
Args: | |
vc_class: a VersionControl subclass. | |
rev: the name of the revision to install. | |
extra_args: a list of extra options. | |
""" | |
if extra_args is None: | |
extra_args = [] | |
self.extra_args = extra_args | |
self.rev = rev | |
self.vc_class = vc_class | |
self.branch_name: Optional[str] = None | |
def __repr__(self) -> str: | |
return f"<RevOptions {self.vc_class.name}: rev={self.rev!r}>" | |
def arg_rev(self) -> Optional[str]: | |
if self.rev is None: | |
return self.vc_class.default_arg_rev | |
return self.rev | |
def to_args(self) -> CommandArgs: | |
""" | |
Return the VCS-specific command arguments. | |
""" | |
args: CommandArgs = [] | |
rev = self.arg_rev | |
if rev is not None: | |
args += self.vc_class.get_base_rev_args(rev) | |
args += self.extra_args | |
return args | |
def to_display(self) -> str: | |
if not self.rev: | |
return "" | |
return f" (to revision {self.rev})" | |
def make_new(self, rev: str) -> "RevOptions": | |
""" | |
Make a copy of the current instance, but with a new rev. | |
Args: | |
rev: the name of the revision for the new object. | |
""" | |
return self.vc_class.make_rev_options(rev, extra_args=self.extra_args) | |
class VcsSupport: | |
_registry: Dict[str, "VersionControl"] = {} | |
schemes = ["ssh", "git", "hg", "bzr", "sftp", "svn"] | |
def __init__(self) -> None: | |
# Register more schemes with urlparse for various version control | |
# systems | |
urllib.parse.uses_netloc.extend(self.schemes) | |
super().__init__() | |
def __iter__(self) -> Iterator[str]: | |
return self._registry.__iter__() | |
def backends(self) -> List["VersionControl"]: | |
return list(self._registry.values()) | |
def dirnames(self) -> List[str]: | |
return [backend.dirname for backend in self.backends] | |
def all_schemes(self) -> List[str]: | |
schemes: List[str] = [] | |
for backend in self.backends: | |
schemes.extend(backend.schemes) | |
return schemes | |
def register(self, cls: Type["VersionControl"]) -> None: | |
if not hasattr(cls, "name"): | |
logger.warning("Cannot register VCS %s", cls.__name__) | |
return | |
if cls.name not in self._registry: | |
self._registry[cls.name] = cls() | |
logger.debug("Registered VCS backend: %s", cls.name) | |
def unregister(self, name: str) -> None: | |
if name in self._registry: | |
del self._registry[name] | |
def get_backend_for_dir(self, location: str) -> Optional["VersionControl"]: | |
""" | |
Return a VersionControl object if a repository of that type is found | |
at the given directory. | |
""" | |
vcs_backends = {} | |
for vcs_backend in self._registry.values(): | |
repo_path = vcs_backend.get_repository_root(location) | |
if not repo_path: | |
continue | |
logger.debug("Determine that %s uses VCS: %s", location, vcs_backend.name) | |
vcs_backends[repo_path] = vcs_backend | |
if not vcs_backends: | |
return None | |
# Choose the VCS in the inner-most directory. Since all repository | |
# roots found here would be either `location` or one of its | |
# parents, the longest path should have the most path components, | |
# i.e. the backend representing the inner-most repository. | |
inner_most_repo_path = max(vcs_backends, key=len) | |
return vcs_backends[inner_most_repo_path] | |
def get_backend_for_scheme(self, scheme: str) -> Optional["VersionControl"]: | |
""" | |
Return a VersionControl object or None. | |
""" | |
for vcs_backend in self._registry.values(): | |
if scheme in vcs_backend.schemes: | |
return vcs_backend | |
return None | |
def get_backend(self, name: str) -> Optional["VersionControl"]: | |
""" | |
Return a VersionControl object or None. | |
""" | |
name = name.lower() | |
return self._registry.get(name) | |
vcs = VcsSupport() | |
class VersionControl: | |
name = "" | |
dirname = "" | |
repo_name = "" | |
# List of supported schemes for this Version Control | |
schemes: Tuple[str, ...] = () | |
# Iterable of environment variable names to pass to call_subprocess(). | |
unset_environ: Tuple[str, ...] = () | |
default_arg_rev: Optional[str] = None | |
def should_add_vcs_url_prefix(cls, remote_url: str) -> bool: | |
""" | |
Return whether the vcs prefix (e.g. "git+") should be added to a | |
repository's remote url when used in a requirement. | |
""" | |
return not remote_url.lower().startswith(f"{cls.name}:") | |
def get_subdirectory(cls, location: str) -> Optional[str]: | |
""" | |
Return the path to Python project root, relative to the repo root. | |
Return None if the project root is in the repo root. | |
""" | |
return None | |
def get_requirement_revision(cls, repo_dir: str) -> str: | |
""" | |
Return the revision string that should be used in a requirement. | |
""" | |
return cls.get_revision(repo_dir) | |
def get_src_requirement(cls, repo_dir: str, project_name: str) -> str: | |
""" | |
Return the requirement string to use to redownload the files | |
currently at the given repository directory. | |
Args: | |
project_name: the (unescaped) project name. | |
The return value has a form similar to the following: | |
{repository_url}@{revision}#egg={project_name} | |
""" | |
repo_url = cls.get_remote_url(repo_dir) | |
if cls.should_add_vcs_url_prefix(repo_url): | |
repo_url = f"{cls.name}+{repo_url}" | |
revision = cls.get_requirement_revision(repo_dir) | |
subdir = cls.get_subdirectory(repo_dir) | |
req = make_vcs_requirement_url(repo_url, revision, project_name, subdir=subdir) | |
return req | |
def get_base_rev_args(rev: str) -> List[str]: | |
""" | |
Return the base revision arguments for a vcs command. | |
Args: | |
rev: the name of a revision to install. Cannot be None. | |
""" | |
raise NotImplementedError | |
def is_immutable_rev_checkout(self, url: str, dest: str) -> bool: | |
""" | |
Return true if the commit hash checked out at dest matches | |
the revision in url. | |
Always return False, if the VCS does not support immutable commit | |
hashes. | |
This method does not check if there are local uncommitted changes | |
in dest after checkout, as pip currently has no use case for that. | |
""" | |
return False | |
def make_rev_options( | |
cls, rev: Optional[str] = None, extra_args: Optional[CommandArgs] = None | |
) -> RevOptions: | |
""" | |
Return a RevOptions object. | |
Args: | |
rev: the name of a revision to install. | |
extra_args: a list of extra options. | |
""" | |
return RevOptions(cls, rev, extra_args=extra_args) | |
def _is_local_repository(cls, repo: str) -> bool: | |
""" | |
posix absolute paths start with os.path.sep, | |
win32 ones start with drive (like c:\\folder) | |
""" | |
drive, tail = os.path.splitdrive(repo) | |
return repo.startswith(os.path.sep) or bool(drive) | |
def get_netloc_and_auth( | |
cls, netloc: str, scheme: str | |
) -> Tuple[str, Tuple[Optional[str], Optional[str]]]: | |
""" | |
Parse the repository URL's netloc, and return the new netloc to use | |
along with auth information. | |
Args: | |
netloc: the original repository URL netloc. | |
scheme: the repository URL's scheme without the vcs prefix. | |
This is mainly for the Subversion class to override, so that auth | |
information can be provided via the --username and --password options | |
instead of through the URL. For other subclasses like Git without | |
such an option, auth information must stay in the URL. | |
Returns: (netloc, (username, password)). | |
""" | |
return netloc, (None, None) | |
def get_url_rev_and_auth(cls, url: str) -> Tuple[str, Optional[str], AuthInfo]: | |
""" | |
Parse the repository URL to use, and return the URL, revision, | |
and auth info to use. | |
Returns: (url, rev, (username, password)). | |
""" | |
scheme, netloc, path, query, frag = urllib.parse.urlsplit(url) | |
if "+" not in scheme: | |
raise ValueError( | |
"Sorry, {!r} is a malformed VCS url. " | |
"The format is <vcs>+<protocol>://<url>, " | |
"e.g. svn+http://myrepo/svn/MyApp#egg=MyApp".format(url) | |
) | |
# Remove the vcs prefix. | |
scheme = scheme.split("+", 1)[1] | |
netloc, user_pass = cls.get_netloc_and_auth(netloc, scheme) | |
rev = None | |
if "@" in path: | |
path, rev = path.rsplit("@", 1) | |
if not rev: | |
raise InstallationError( | |
"The URL {!r} has an empty revision (after @) " | |
"which is not supported. Include a revision after @ " | |
"or remove @ from the URL.".format(url) | |
) | |
url = urllib.parse.urlunsplit((scheme, netloc, path, query, "")) | |
return url, rev, user_pass | |
def make_rev_args( | |
username: Optional[str], password: Optional[HiddenText] | |
) -> CommandArgs: | |
""" | |
Return the RevOptions "extra arguments" to use in obtain(). | |
""" | |
return [] | |
def get_url_rev_options(self, url: HiddenText) -> Tuple[HiddenText, RevOptions]: | |
""" | |
Return the URL and RevOptions object to use in obtain(), | |
as a tuple (url, rev_options). | |
""" | |
secret_url, rev, user_pass = self.get_url_rev_and_auth(url.secret) | |
username, secret_password = user_pass | |
password: Optional[HiddenText] = None | |
if secret_password is not None: | |
password = hide_value(secret_password) | |
extra_args = self.make_rev_args(username, password) | |
rev_options = self.make_rev_options(rev, extra_args=extra_args) | |
return hide_url(secret_url), rev_options | |
def normalize_url(url: str) -> str: | |
""" | |
Normalize a URL for comparison by unquoting it and removing any | |
trailing slash. | |
""" | |
return urllib.parse.unquote(url).rstrip("/") | |
def compare_urls(cls, url1: str, url2: str) -> bool: | |
""" | |
Compare two repo URLs for identity, ignoring incidental differences. | |
""" | |
return cls.normalize_url(url1) == cls.normalize_url(url2) | |
def fetch_new( | |
self, dest: str, url: HiddenText, rev_options: RevOptions, verbosity: int | |
) -> None: | |
""" | |
Fetch a revision from a repository, in the case that this is the | |
first fetch from the repository. | |
Args: | |
dest: the directory to fetch the repository to. | |
rev_options: a RevOptions object. | |
verbosity: verbosity level. | |
""" | |
raise NotImplementedError | |
def switch(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None: | |
""" | |
Switch the repo at ``dest`` to point to ``URL``. | |
Args: | |
rev_options: a RevOptions object. | |
""" | |
raise NotImplementedError | |
def update(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None: | |
""" | |
Update an already-existing repo to the given ``rev_options``. | |
Args: | |
rev_options: a RevOptions object. | |
""" | |
raise NotImplementedError | |
def is_commit_id_equal(cls, dest: str, name: Optional[str]) -> bool: | |
""" | |
Return whether the id of the current commit equals the given name. | |
Args: | |
dest: the repository directory. | |
name: a string name. | |
""" | |
raise NotImplementedError | |
def obtain(self, dest: str, url: HiddenText, verbosity: int) -> None: | |
""" | |
Install or update in editable mode the package represented by this | |
VersionControl object. | |
:param dest: the repository directory in which to install or update. | |
:param url: the repository URL starting with a vcs prefix. | |
:param verbosity: verbosity level. | |
""" | |
url, rev_options = self.get_url_rev_options(url) | |
if not os.path.exists(dest): | |
self.fetch_new(dest, url, rev_options, verbosity=verbosity) | |
return | |
rev_display = rev_options.to_display() | |
if self.is_repository_directory(dest): | |
existing_url = self.get_remote_url(dest) | |
if self.compare_urls(existing_url, url.secret): | |
logger.debug( | |
"%s in %s exists, and has correct URL (%s)", | |
self.repo_name.title(), | |
display_path(dest), | |
url, | |
) | |
if not self.is_commit_id_equal(dest, rev_options.rev): | |
logger.info( | |
"Updating %s %s%s", | |
display_path(dest), | |
self.repo_name, | |
rev_display, | |
) | |
self.update(dest, url, rev_options) | |
else: | |
logger.info("Skipping because already up-to-date.") | |
return | |
logger.warning( | |
"%s %s in %s exists with URL %s", | |
self.name, | |
self.repo_name, | |
display_path(dest), | |
existing_url, | |
) | |
prompt = ("(s)witch, (i)gnore, (w)ipe, (b)ackup ", ("s", "i", "w", "b")) | |
else: | |
logger.warning( | |
"Directory %s already exists, and is not a %s %s.", | |
dest, | |
self.name, | |
self.repo_name, | |
) | |
# https://github.com/python/mypy/issues/1174 | |
prompt = ("(i)gnore, (w)ipe, (b)ackup ", ("i", "w", "b")) # type: ignore | |
logger.warning( | |
"The plan is to install the %s repository %s", | |
self.name, | |
url, | |
) | |
response = ask_path_exists("What to do? {}".format(prompt[0]), prompt[1]) | |
if response == "a": | |
sys.exit(-1) | |
if response == "w": | |
logger.warning("Deleting %s", display_path(dest)) | |
rmtree(dest) | |
self.fetch_new(dest, url, rev_options, verbosity=verbosity) | |
return | |
if response == "b": | |
dest_dir = backup_dir(dest) | |
logger.warning("Backing up %s to %s", display_path(dest), dest_dir) | |
shutil.move(dest, dest_dir) | |
self.fetch_new(dest, url, rev_options, verbosity=verbosity) | |
return | |
# Do nothing if the response is "i". | |
if response == "s": | |
logger.info( | |
"Switching %s %s to %s%s", | |
self.repo_name, | |
display_path(dest), | |
url, | |
rev_display, | |
) | |
self.switch(dest, url, rev_options) | |
def unpack(self, location: str, url: HiddenText, verbosity: int) -> None: | |
""" | |
Clean up current location and download the url repository | |
(and vcs infos) into location | |
:param url: the repository URL starting with a vcs prefix. | |
:param verbosity: verbosity level. | |
""" | |
if os.path.exists(location): | |
rmtree(location) | |
self.obtain(location, url=url, verbosity=verbosity) | |
def get_remote_url(cls, location: str) -> str: | |
""" | |
Return the url used at location | |
Raises RemoteNotFoundError if the repository does not have a remote | |
url configured. | |
""" | |
raise NotImplementedError | |
def get_revision(cls, location: str) -> str: | |
""" | |
Return the current commit id of the files at the given location. | |
""" | |
raise NotImplementedError | |
def run_command( | |
cls, | |
cmd: Union[List[str], CommandArgs], | |
show_stdout: bool = True, | |
cwd: Optional[str] = None, | |
on_returncode: 'Literal["raise", "warn", "ignore"]' = "raise", | |
extra_ok_returncodes: Optional[Iterable[int]] = None, | |
command_desc: Optional[str] = None, | |
extra_environ: Optional[Mapping[str, Any]] = None, | |
spinner: Optional[SpinnerInterface] = None, | |
log_failed_cmd: bool = True, | |
stdout_only: bool = False, | |
) -> str: | |
""" | |
Run a VCS subcommand | |
This is simply a wrapper around call_subprocess that adds the VCS | |
command name, and checks that the VCS is available | |
""" | |
cmd = make_command(cls.name, *cmd) | |
if command_desc is None: | |
command_desc = format_command_args(cmd) | |
try: | |
return call_subprocess( | |
cmd, | |
show_stdout, | |
cwd, | |
on_returncode=on_returncode, | |
extra_ok_returncodes=extra_ok_returncodes, | |
command_desc=command_desc, | |
extra_environ=extra_environ, | |
unset_environ=cls.unset_environ, | |
spinner=spinner, | |
log_failed_cmd=log_failed_cmd, | |
stdout_only=stdout_only, | |
) | |
except FileNotFoundError: | |
# errno.ENOENT = no such file or directory | |
# In other words, the VCS executable isn't available | |
raise BadCommand( | |
f"Cannot find command {cls.name!r} - do you have " | |
f"{cls.name!r} installed and in your PATH?" | |
) | |
except PermissionError: | |
# errno.EACCES = Permission denied | |
# This error occurs, for instance, when the command is installed | |
# only for another user. So, the current user don't have | |
# permission to call the other user command. | |
raise BadCommand( | |
f"No permission to execute {cls.name!r} - install it " | |
f"locally, globally (ask admin), or check your PATH. " | |
f"See possible solutions at " | |
f"https://pip.pypa.io/en/latest/reference/pip_freeze/" | |
f"#fixing-permission-denied." | |
) | |
def is_repository_directory(cls, path: str) -> bool: | |
""" | |
Return whether a directory path is a repository directory. | |
""" | |
logger.debug("Checking in %s for %s (%s)...", path, cls.dirname, cls.name) | |
return os.path.exists(os.path.join(path, cls.dirname)) | |
def get_repository_root(cls, location: str) -> Optional[str]: | |
""" | |
Return the "root" (top-level) directory controlled by the vcs, | |
or `None` if the directory is not in any. | |
It is meant to be overridden to implement smarter detection | |
mechanisms for specific vcs. | |
This can do more than is_repository_directory() alone. For | |
example, the Git override checks that Git is actually available. | |
""" | |
if cls.is_repository_directory(location): | |
return location | |
return None | |