juc500 정상동작

This commit is contained in:
Macbook
2026-07-21 21:04:25 +09:00
parent ff60cee1cc
commit 1388668b56
2917 changed files with 708021 additions and 651 deletions
@@ -0,0 +1,169 @@
from __future__ import annotations
import logging
from abc import ABC
from argparse import SUPPRESS
from pathlib import Path
from typing import TYPE_CHECKING
from virtualenv.seed.seeder import Seeder
from virtualenv.seed.wheels import Version
from virtualenv.seed.wheels.embed import MIN, OLDEST_SUPPORTED
if TYPE_CHECKING:
from argparse import ArgumentParser
from python_discovery import PythonInfo
from virtualenv.app_data.base import AppData
from virtualenv.config.cli.parser import VirtualEnvOptions
LOGGER = logging.getLogger(__name__)
PERIODIC_UPDATE_ON_BY_DEFAULT = True
class BaseEmbed(Seeder, ABC):
def __init__(self, options: VirtualEnvOptions) -> None:
super().__init__(options, enabled=options.no_seed is False)
self.download = options.download
self.extra_search_dir = [i.resolve() for i in options.extra_search_dir if i.exists()]
self.pip_version = options.pip
self.setuptools_version = options.setuptools
# virtualenv no longer bundles wheel; the parsed default stays None (unused) so the
# warning below fires only when you pass --wheel or --no-wheel
self.wheel_version = options.wheel or "none"
self.no_pip = options.no_pip
self.no_setuptools = options.no_setuptools
self.app_data = options.app_data
self.periodic_update = not options.no_periodic_update
if options.wheel is not None or options.no_wheel:
LOGGER.warning(
"DEPRECATION: the --wheel and --no-wheel options do nothing; virtualenv no longer bundles wheel. "
"They will be removed in a release after 2026-12. Stop passing them.",
)
self.no_wheel = True
if not self.distribution_to_versions():
self.enabled = False
@classmethod
def distributions(cls) -> dict[str, str]:
return {
"pip": Version.bundle,
"setuptools": Version.bundle,
"wheel": Version.bundle,
}
def distribution_to_versions(self) -> dict[str, str]:
return {
distribution: getattr(self, f"{distribution}_version")
for distribution in self.distributions()
if getattr(self, f"no_{distribution}", None) is False and getattr(self, f"{distribution}_version") != "none"
}
@classmethod
def cannot_seed(cls, interpreter: PythonInfo) -> str | None:
"""Explain why the bundled wheels cannot seed the target Python version.
The embedded pip/setuptools stopped shipping for Pythons below :data:`OLDEST_SUPPORTED`, so seeding one would
install an incompatible wheel.
:param interpreter: the interpreter to be seeded
:returns: ``None`` when the bundled wheels still support the target, otherwise a message naming the target and
the remedies
"""
if interpreter.version_info[:2] >= OLDEST_SUPPORTED:
return None
target = f"{interpreter.version_info.major}.{interpreter.version_info.minor}"
return (
f"the bundled seeder no longer ships pip/setuptools for Python {target}; the oldest supported target is "
f"Python {MIN} - pass --no-seed for an empty environment, use a seeder that provides Python {target} "
f"wheels, or install an older virtualenv release"
)
@classmethod
def add_parser_arguments(cls, parser: ArgumentParser, interpreter: PythonInfo, app_data: AppData) -> None: # noqa: ARG003
group = parser.add_mutually_exclusive_group()
group.add_argument(
"--no-download",
"--never-download",
dest="download",
action="store_false",
help=f"pass to disable download of the latest {'/'.join(cls.distributions())} from PyPI",
default=True,
)
group.add_argument(
"--download",
dest="download",
action="store_true",
help=f"pass to enable download of the latest {'/'.join(cls.distributions())} from PyPI",
default=False,
)
parser.add_argument(
"--extra-search-dir",
metavar="d",
type=Path,
nargs="+",
help="a path containing wheels to extend the internal wheel list (can be set 1+ times)",
default=[],
)
for distribution, default in cls.distributions().items():
help_ = f"version of {distribution} to install as seed: embed, bundle, none or exact version"
if interpreter.version_info[:2] >= (3, 12) and distribution == "setuptools":
default = "none" # noqa: PLW2901
if distribution == "wheel":
default = None # noqa: PLW2901
help_ = SUPPRESS
parser.add_argument(
f"--{distribution}",
dest=distribution,
metavar="version",
help=help_,
default=default,
)
for distribution in cls.distributions():
help_ = f"do not install {distribution}"
if distribution == "wheel":
help_ = SUPPRESS
parser.add_argument(
f"--no-{distribution}",
dest=f"no_{distribution}",
action="store_true",
help=help_,
default=False,
)
parser.add_argument(
"--no-periodic-update",
dest="no_periodic_update",
action="store_true",
help="disable the periodic (once every 14 days) update of the embedded wheels",
default=not PERIODIC_UPDATE_ON_BY_DEFAULT,
)
def __repr__(self) -> str:
result = self.__class__.__name__
result += "("
if self.extra_search_dir:
result += f"extra_search_dir={', '.join(str(i) for i in self.extra_search_dir)},"
result += f"download={self.download},"
for distribution in self.distributions():
if getattr(self, f"no_{distribution}", None):
continue
version = getattr(self, f"{distribution}_version", None)
if version == "none":
continue
ver = f"={version or 'latest'}"
result += f" {distribution}{ver},"
return result[:-1] + ")"
__all__ = [
"BaseEmbed",
]
@@ -0,0 +1,83 @@
from __future__ import annotations
import logging
from contextlib import contextmanager
from subprocess import Popen
from typing import TYPE_CHECKING
from virtualenv.seed.embed.base_embed import BaseEmbed
from virtualenv.seed.wheels import Version, get_wheel, pip_wheel_env_run
from virtualenv.util.subprocess import LogCmd
if TYPE_CHECKING:
from collections.abc import Generator
from pathlib import Path
from virtualenv.config.cli.parser import VirtualEnvOptions
from virtualenv.create.creator import Creator
LOGGER = logging.getLogger(__name__)
class PipInvoke(BaseEmbed):
def __init__(self, options: VirtualEnvOptions) -> None:
super().__init__(options)
def run(self, creator: Creator) -> None:
if not self.enabled:
return
for_py_version = creator.interpreter.version_release_str
with self.get_pip_install_cmd(creator.exe, for_py_version) as cmd:
env = pip_wheel_env_run(self.extra_search_dir, self.app_data, self.env)
self._execute(cmd, env)
@staticmethod
def _execute(cmd: list[str], env: dict[str, str]) -> Popen[bytes]:
LOGGER.debug("pip seed by running: %s", LogCmd(cmd, env))
process = Popen(cmd, env=env)
process.communicate()
if process.returncode != 0:
msg = f"failed seed with code {process.returncode}"
raise RuntimeError(msg)
return process
@contextmanager
def get_pip_install_cmd(self, exe: Path, for_py_version: str) -> Generator[list[str], None, None]:
cmd = [
str(exe),
"-m",
"pip",
"-q",
"install",
"--only-binary",
":all:",
"--disable-pip-version-check",
"--ignore-installed",
]
if not self.download:
cmd.append("--no-index")
folders = set()
for dist, version in self.distribution_to_versions().items():
wheel = get_wheel(
distribution=dist,
version=version,
for_py_version=for_py_version,
search_dirs=self.extra_search_dir,
download=False,
app_data=self.app_data,
do_periodic_update=self.periodic_update,
env=self.env,
)
if wheel is None:
msg = f"could not get wheel for distribution {dist}"
raise RuntimeError(msg)
folders.add(str(wheel.path.parent))
cmd.append(Version.as_pip_req(dist, wheel.version))
for folder in sorted(folders):
cmd.extend(["--find-links", str(folder)])
yield cmd
__all__ = [
"PipInvoke",
]
@@ -0,0 +1,242 @@
from __future__ import annotations
import logging
import ntpath
import os
import posixpath
import re
import zipfile
from abc import ABC, abstractmethod
from configparser import ConfigParser
from itertools import chain
from pathlib import Path
from tempfile import mkdtemp
from typing import TYPE_CHECKING
from distlib.scripts import ScriptMaker, enquote_executable
from virtualenv.util.path import safe_delete
if TYPE_CHECKING:
from virtualenv.create.creator import Creator
LOGGER = logging.getLogger(__name__)
def _safe_extract_zip(zip_ref: zipfile.ZipFile, target_dir: Path) -> None:
# Guard against zip slip: a wheel is a zip and a tampered entry name (absolute path or one containing ``..``)
# could escape ``target_dir``.
base = target_dir.resolve()
for info in zip_ref.infolist():
name = info.filename
if name.startswith(("/", "\\")) or ntpath.isabs(name) or posixpath.isabs(name):
msg = f"refusing to extract absolute path entry from wheel: {name!r}"
raise RuntimeError(msg)
candidate = (base / name).resolve()
try:
candidate.relative_to(base)
except ValueError as exc:
msg = f"refusing to extract entry escaping target directory: {name!r}"
raise RuntimeError(msg) from exc
zip_ref.extractall(str(target_dir))
class PipInstall(ABC):
def __init__(self, wheel: Path, creator: Creator, image_folder: Path) -> None:
self._wheel = wheel
self._creator = creator
self._image_dir = image_folder
self._extracted = False
self.__dist_info = None
self._console_entry_points = None
@abstractmethod
def _sync(self, src: Path, dst: Path) -> None:
raise NotImplementedError
def install(self, version_info: tuple[int, ...]) -> None:
self._extracted = True
self._uninstall_previous_version()
# sync image
for filename in self._image_dir.iterdir():
into = self._creator.purelib / filename.name
self._sync(filename, into)
# generate console executables
consoles = set()
script_dir = self._creator.script_dir
for name, module in self._console_scripts.items(): # ty: ignore[unresolved-attribute]
consoles.update(self._create_console_entry_point(name, module, script_dir, version_info))
LOGGER.debug("generated console scripts %s", " ".join(i.name for i in consoles))
def build_image(self) -> None:
"""Extract the seed wheel into the image directory and fix up its RECORD file.
Each archive entry is validated before extraction so a tampered wheel cannot escape the image directory via an
absolute path or ``..`` traversal.
:raises RuntimeError: if the wheel contains an entry that would land outside the image directory.
"""
# 1. first extract the wheel
LOGGER.debug("build install image for %s to %s", self._wheel.name, self._image_dir)
with zipfile.ZipFile(str(self._wheel)) as zip_ref:
self._shorten_path_if_needed(zip_ref)
_safe_extract_zip(zip_ref, self._image_dir)
self._extracted = True
# 2. now add additional files not present in the distribution
new_files = self._generate_new_files()
# 3. finally fix the records file
self._fix_records(new_files)
def _shorten_path_if_needed(self, zip_ref: zipfile.ZipFile) -> None:
if os.name == "nt":
to_folder = str(self._image_dir)
# https://docs.microsoft.com/en-us/windows/win32/fileio/maximum-file-path-limitation
zip_max_len = max(len(i) for i in zip_ref.namelist())
path_len = zip_max_len + len(to_folder)
if path_len > 260: # noqa: PLR2004
self._image_dir.mkdir(exist_ok=True) # to get a short path must exist
from virtualenv.util.path import get_short_path_name # noqa: PLC0415
to_folder = get_short_path_name(to_folder)
self._image_dir = Path(to_folder)
def _records_text(self, files: set[Path] | list[Path]) -> str:
return "\n".join(f"{os.path.relpath(str(rec), str(self._image_dir))},," for rec in files)
def _generate_new_files(self) -> set[Path]:
new_files = set()
installer = self._dist_info / "INSTALLER" # ty: ignore[unsupported-operator]
installer.write_text("pip\n", encoding="utf-8")
new_files.add(installer)
# inject a no-op root element, as workaround for bug in https://github.com/pypa/pip/issues/7226
marker = self._image_dir / f"{self._dist_info.stem}.virtualenv" # ty: ignore[unresolved-attribute]
marker.write_text("", encoding="utf-8")
new_files.add(marker)
folder = mkdtemp()
try:
to_folder = Path(folder)
rel = os.path.relpath(str(self._creator.script_dir), str(self._creator.purelib))
version_info = self._creator.interpreter.version_info
for name, module in self._console_scripts.items(): # ty: ignore[unresolved-attribute]
new_files.update(
Path(os.path.normpath(str(self._image_dir / rel / i.name)))
for i in self._create_console_entry_point(name, module, to_folder, version_info) # ty: ignore[invalid-argument-type]
)
finally:
safe_delete(folder) # ty: ignore[invalid-argument-type]
return new_files
@property
def _dist_info(self) -> Path | None:
if self._extracted is False:
return None # pragma: no cover
if self.__dist_info is None:
files = []
for filename in self._image_dir.iterdir():
files.append(filename.name)
if filename.suffix == ".dist-info":
self.__dist_info = filename
break
else:
msg = f"no .dist-info at {self._image_dir}, has {', '.join(files)}"
raise RuntimeError(msg) # pragma: no cover
return self.__dist_info
@abstractmethod
def _fix_records(self, extra_record_data: set[Path]) -> None:
raise NotImplementedError
@property
def _console_scripts(self) -> dict[str, str] | None:
if self._extracted is False:
return None # pragma: no cover
if self._console_entry_points is None:
self._console_entry_points = {}
entry_points = self._dist_info / "entry_points.txt" # ty: ignore[unsupported-operator]
if entry_points.exists():
parser = ConfigParser()
with entry_points.open(encoding="utf-8") as file_handler:
parser.read_file(file_handler)
if "console_scripts" in parser.sections():
for name, value in parser.items("console_scripts"):
match = re.match(r"(.*?)-?\d\.?\d*", name)
our_name = match.group(1) if match else name
self._console_entry_points[our_name] = value
return self._console_entry_points
def _create_console_entry_point(
self, name: str, value: str, to_folder: Path, version_info: tuple[int, ...]
) -> list[Path]:
result = []
maker = ScriptMakerCustom(to_folder, version_info, self._creator.exe, name)
specification = f"{name} = {value}"
new_files = maker.make(specification)
result.extend(Path(i) for i in new_files)
return result
def _uninstall_previous_version(self) -> None:
dist_name = self._dist_info.stem.split("-")[0] # ty: ignore[unresolved-attribute]
in_folders = chain.from_iterable([i.iterdir() for i in (self._creator.purelib, self._creator.platlib)])
paths = (p for p in in_folders if p.stem.split("-")[0] == dist_name and p.suffix == ".dist-info" and p.is_dir())
existing_dist = next(paths, None)
if existing_dist is not None:
self._uninstall_dist(existing_dist)
@staticmethod
def _uninstall_dist(dist: Path) -> None:
dist_base = dist.parent
LOGGER.debug("uninstall existing distribution %s from %s", dist.stem, dist_base)
top_txt = dist / "top_level.txt" # add top level packages at folder level
paths = (
{dist.parent / i.strip() for i in top_txt.read_text(encoding="utf-8").splitlines()}
if top_txt.exists()
else set()
)
paths.add(dist) # add the dist-info folder itself
base_dirs, record = paths.copy(), dist / "RECORD" # collect entries in record that we did not register yet
for name in (
(i.split(",")[0] for i in record.read_text(encoding="utf-8").splitlines()) if record.exists() else ()
):
path = dist_base / name
if not any(p in base_dirs for p in path.parents): # only add if not already added as a base dir
paths.add(path)
for path in sorted(paths): # actually remove stuff in a stable order
if path.exists():
if path.is_dir() and not path.is_symlink():
safe_delete(path)
else:
path.unlink()
def clear(self) -> None:
if self._image_dir.exists():
safe_delete(self._image_dir)
def has_image(self) -> bool:
return self._image_dir.exists() and any(self._image_dir.iterdir())
class ScriptMakerCustom(ScriptMaker):
def __init__(self, target_dir: Path, version_info: tuple[int, ...], executable: Path, name: str) -> None:
super().__init__(None, str(target_dir))
self.clobber = True # overwrite
self.set_mode = True # ensure they are executable
self.executable = enquote_executable(str(executable))
self.version_info = version_info.major, version_info.minor # ty: ignore[unresolved-attribute]
self.variants = {"", "X", "X.Y"}
self._name = name
def _write_script(
self, names: set[str], shebang: bytes, script_bytes: bytes, filenames: list[str], ext: str
) -> None:
names.add(f"{self._name}{self.version_info[0]}.{self.version_info[1]}")
super()._write_script(names, shebang, script_bytes, filenames, ext)
__all__ = [
"PipInstall",
]
@@ -0,0 +1,44 @@
from __future__ import annotations
import os
from pathlib import Path
from typing import TYPE_CHECKING
from virtualenv.util.path import copy
from .base import PipInstall
if TYPE_CHECKING:
from collections.abc import Generator
class CopyPipInstall(PipInstall):
def _sync(self, src: Path, dst: Path) -> None:
copy(src, dst)
def _generate_new_files(self) -> set[Path]:
# create the pyc files
new_files = super()._generate_new_files()
new_files.update(self._cache_files())
return new_files
def _cache_files(self) -> Generator[Path, None, None]:
version = self._creator.interpreter.version_info
py_c_ext = f".{self._creator.interpreter.implementation.lower()}-{version.major}{version.minor}.pyc"
for root, dirs, files in os.walk(str(self._image_dir), topdown=True):
root_path = Path(root)
for name in files:
if name.endswith(".py"):
yield root_path / f"{name[:-3]}{py_c_ext}"
for name in dirs:
yield root_path / name / "__pycache__"
def _fix_records(self, extra_record_data: set[Path]) -> None:
extra_record_data_str = self._records_text(extra_record_data)
with (self._dist_info / "RECORD").open("ab") as file_handler: # ty: ignore[unsupported-operator]
file_handler.write(extra_record_data_str.encode("utf-8"))
__all__ = [
"CopyPipInstall",
]
@@ -0,0 +1,62 @@
from __future__ import annotations
import os
from stat import S_IREAD, S_IRGRP, S_IROTH
from subprocess import PIPE, Popen
from typing import TYPE_CHECKING
from virtualenv.util.path import safe_delete, set_tree
from .base import PipInstall
if TYPE_CHECKING:
from pathlib import Path
class SymlinkPipInstall(PipInstall):
def _sync(self, src: Path, dst: Path) -> None:
os.symlink(str(src), str(dst))
def _generate_new_files(self) -> set[Path]:
# create the pyc files, as the build image will be R/O
cmd = [str(self._creator.exe), "-m", "compileall", str(self._image_dir)]
process = Popen(cmd, stdout=PIPE, stderr=PIPE)
process.communicate()
# the root pyc is shared, so we'll not symlink that - but still add the pyc files to the RECORD for close
root_py_cache = self._image_dir / "__pycache__"
new_files = set()
if root_py_cache.exists():
new_files.update(root_py_cache.iterdir())
new_files.add(root_py_cache)
safe_delete(root_py_cache)
core_new_files = super()._generate_new_files()
# remove files that are within the image folder deeper than one level (as these will be not linked directly)
for file in core_new_files:
try:
rel = file.relative_to(self._image_dir)
if len(rel.parts) > 1:
continue
except ValueError:
pass
new_files.add(file)
return new_files
def _fix_records(self, extra_record_data: set[Path]) -> None:
extra_record_data.update(i for i in self._image_dir.iterdir())
extra_record_data_str = self._records_text(sorted(extra_record_data, key=str)) # ty: ignore[invalid-argument-type]
(self._dist_info / "RECORD").write_text(extra_record_data_str, encoding="utf-8") # ty: ignore[unsupported-operator]
def build_image(self) -> None:
super().build_image()
# protect the image by making it read only
set_tree(self._image_dir, S_IREAD | S_IRGRP | S_IROTH)
def clear(self) -> None:
if self._image_dir.exists():
safe_delete(self._image_dir)
super().clear()
__all__ = [
"SymlinkPipInstall",
]
@@ -0,0 +1,164 @@
"""Bootstrap."""
from __future__ import annotations
import logging
import sys
import traceback
from contextlib import contextmanager
from pathlib import Path
from subprocess import CalledProcessError
from threading import Lock, Thread
from typing import TYPE_CHECKING
from virtualenv.info import fs_supports_symlink
from virtualenv.seed.embed.base_embed import BaseEmbed
from virtualenv.seed.wheels import get_wheel
from .pip_install.copy import CopyPipInstall
from .pip_install.symlink import SymlinkPipInstall
if TYPE_CHECKING:
from argparse import ArgumentParser
from collections.abc import Generator
from python_discovery import PythonInfo
from virtualenv.app_data.base import AppData
from virtualenv.config.cli.parser import VirtualEnvOptions
from virtualenv.create.creator import Creator
from virtualenv.seed.wheels.util import Wheel
from virtualenv.util.lock import PathLockBase
from .pip_install.base import PipInstall
LOGGER = logging.getLogger(__name__)
class FromAppData(BaseEmbed):
def __init__(self, options: VirtualEnvOptions) -> None:
super().__init__(options)
self.symlinks = options.symlink_app_data
@classmethod
def add_parser_arguments(cls, parser: ArgumentParser, interpreter: PythonInfo, app_data: AppData) -> None:
super().add_parser_arguments(parser, interpreter, app_data)
can_symlink = app_data.transient is False and fs_supports_symlink()
sym = "" if can_symlink else "not supported - "
parser.add_argument(
"--symlink-app-data",
dest="symlink_app_data",
action="store_true" if can_symlink else "store_false",
help=f"{sym} symlink the python packages from the app-data folder (requires seed pip>=19.3)",
default=False,
)
def run(self, creator: Creator) -> None:
if not self.enabled:
return
with self._get_seed_wheels(creator) as name_to_whl:
pip_version = name_to_whl["pip"].version_tuple if "pip" in name_to_whl else None
installer_class = self.installer_class(pip_version)
exceptions = {}
def _install(name: str, wheel: Wheel) -> None:
LOGGER.debug("install %s from wheel %s via %s", name, wheel, installer_class.__name__)
try:
key = Path(installer_class.__name__) / wheel.path.stem
wheel_img = self.app_data.wheel_image(creator.interpreter.version_release_str, key)
installer = installer_class(wheel.path, creator, wheel_img)
_build_wheel_image(self.app_data.lock / wheel_img.parent, wheel_img.name, installer)
installer.install(creator.interpreter.version_info) # ty: ignore[invalid-argument-type]
except Exception: # noqa: BLE001
exceptions[name] = sys.exc_info()
threads = [Thread(target=_install, args=(n, w)) for n, w in name_to_whl.items()]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
if exceptions:
messages = [f"failed to build image {', '.join(exceptions.keys())} because:"]
for value in exceptions.values():
exc_type, exc_value, exc_traceback = value
messages.append("".join(traceback.format_exception(exc_type, exc_value, exc_traceback)))
raise RuntimeError("\n".join(messages))
@contextmanager
def _get_seed_wheels(self, creator: Creator) -> Generator[dict[str, Wheel], None, None]: # noqa: C901
name_to_whl, lock, fail = {}, Lock(), {}
def _get(distribution: str, version: str | None) -> None:
for_py_version = creator.interpreter.version_release_str
failure, result = None, None
# fallback to download in case the exact version is not available
for download in [True] if self.download else [False, True]:
failure = None
try:
result = get_wheel(
distribution=distribution,
version=version,
for_py_version=for_py_version,
search_dirs=self.extra_search_dir,
download=download,
app_data=self.app_data,
do_periodic_update=self.periodic_update,
env=self.env,
)
if result is not None:
break
except Exception as exception:
LOGGER.exception("fail")
failure = exception
if failure:
if isinstance(failure, CalledProcessError):
msg = f"failed to download {distribution}"
if version is not None:
msg += f" version {version}"
msg += f", pip download exit code {failure.returncode}"
output = failure.output + failure.stderr
if output:
msg += "\n"
msg += output
else:
msg = repr(failure)
LOGGER.error(msg)
with lock:
fail[distribution] = version
else:
with lock:
name_to_whl[distribution] = result
threads = [
Thread(target=_get, args=(distribution, version))
for distribution, version in self.distribution_to_versions().items()
]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
if fail:
msg = f"seed failed due to failing to download wheels {', '.join(fail.keys())}"
raise RuntimeError(msg)
yield name_to_whl
def installer_class(self, pip_version_tuple: tuple[int, ...] | None) -> type[PipInstall]:
if self.symlinks and pip_version_tuple and pip_version_tuple >= (19, 3): # symlink support requires pip 19.3+
return SymlinkPipInstall
return CopyPipInstall
def __repr__(self) -> str:
msg = f", via={'symlink' if self.symlinks else 'copy'}, app_data_dir={self.app_data}"
base = super().__repr__()
return f"{base[:-1]}{msg}{base[-1]}"
def _build_wheel_image(parent: PathLockBase, name: str, installer: PipInstall) -> None:
with parent.non_reentrant_lock_for_key(name):
if not installer.has_image():
installer.build_image()
__all__ = [
"FromAppData",
]
@@ -0,0 +1,65 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from argparse import ArgumentParser
from python_discovery import PythonInfo
from virtualenv.app_data.base import AppData
from virtualenv.config.cli.parser import VirtualEnvOptions
from virtualenv.create.creator import Creator
class Seeder(ABC):
"""A seeder will install some seed packages into a virtual environment."""
def __init__(self, options: VirtualEnvOptions, enabled: bool) -> None:
"""Create.
:param options: the parsed options as defined within :meth:`add_parser_arguments`
:param enabled: a flag weather the seeder is enabled or not
"""
self.enabled = enabled
self.env = options.env
@classmethod
def cannot_seed(cls, interpreter: PythonInfo) -> str | None: # noqa: ARG003
"""Explain why this seeder cannot install seed packages for the given interpreter.
:param interpreter: the interpreter the environment is based on
:returns: ``None`` when the seeder supports the interpreter, otherwise a message describing why it cannot;
selection rejects a seeder that returns a message and surfaces it to the user
"""
return None
@classmethod
def add_parser_arguments(cls, parser: ArgumentParser, interpreter: PythonInfo, app_data: AppData) -> None:
"""Add CLI arguments for this seed mechanisms.
:param parser: the CLI parser
:param app_data: the CLI parser
:param interpreter: the interpreter this virtual environment is based of
"""
raise NotImplementedError
@abstractmethod
def run(self, creator: Creator) -> None:
"""Perform the seed operation.
:param creator: the creator (based of :class:`virtualenv.create.creator.Creator`) we used to create this virtual
environment
"""
raise NotImplementedError
__all__ = [
"Seeder",
]
@@ -0,0 +1,11 @@
from __future__ import annotations
from .acquire import get_wheel, pip_wheel_env_run
from .util import Version, Wheel
__all__ = [
"Version",
"Wheel",
"get_wheel",
"pip_wheel_env_run",
]
@@ -0,0 +1,213 @@
"""Bootstrap."""
from __future__ import annotations
import logging
import re
import sys
from operator import eq, lt
from pathlib import Path
from subprocess import PIPE, CalledProcessError, Popen
from typing import TYPE_CHECKING
from .bundle import from_bundle
from .periodic_update import add_wheel_to_update_log
from .util import Version, Wheel, discover_wheels
if TYPE_CHECKING:
from virtualenv.app_data.base import AppData
LOGGER = logging.getLogger(__name__)
# PEP 503 normalized distribution name. Anything outside this character set on the way to ``pip download`` means
# somebody is smuggling pip options or extras, so reject it before we build the command line.
_DISTRIBUTION_RE = re.compile(
r"""
^
(?P<name>
[A-Za-z0-9] # must start with an alnum
(?:[A-Za-z0-9._-]* # inner chars: alnum plus . _ -
[A-Za-z0-9])? # must also end with an alnum (unless length is 1)
)
$
""",
re.VERBOSE,
)
# Version specifier that matches what ``Version.as_version_spec`` emits: either empty, ``==<ver>`` or ``<<ver>`` where
# ``<ver>`` is a subset of PEP 440 public versions. Kept deliberately strict so a crafted version cannot inject pip
# flags.
_VERSION_SPEC_RE = re.compile(
r"""
^
(?P<operator>==|<) # only the operators Version.as_version_spec can emit
(?P<version>[A-Za-z0-9._+!-]+) # PEP 440 public-version character set, no whitespace
$
""",
re.VERBOSE,
)
def get_wheel( # noqa: PLR0913
distribution: str,
version: str | None,
for_py_version: str,
search_dirs: list[Path],
download: bool,
app_data: AppData,
do_periodic_update: bool,
env: dict[str, str],
) -> Wheel | None:
"""Get a wheel with the given distribution-version-for_py_version trio, by using the extra search dir + download."""
# not all wheels are compatible with all python versions, so we need to py version qualify it
wheel = None
if not download or version != Version.bundle:
# 1. acquire from bundle
wheel = from_bundle(distribution, version, for_py_version, search_dirs, app_data, do_periodic_update, env)
if download and wheel is None and version != Version.embed:
# 2. download from the internet
wheel = download_wheel(
distribution=distribution,
version_spec=Version.as_version_spec(version),
for_py_version=for_py_version,
search_dirs=search_dirs,
app_data=app_data,
to_folder=app_data.house,
env=env,
)
if wheel is not None and app_data.can_update:
add_wheel_to_update_log(wheel, for_py_version, app_data)
return wheel
def download_wheel( # noqa: PLR0913
distribution: str,
version_spec: str | None,
for_py_version: str,
search_dirs: list[Path],
app_data: AppData,
to_folder: Path,
env: dict[str, str],
) -> Wheel:
"""Invoke ``pip download`` in a subprocess to fetch a seed wheel.
:param distribution: PEP 503 normalized project name; rejected if it contains anything other than
``[A-Za-z0-9._-]``.
:param version_spec: optional version specifier of the form ``==<ver>`` or ``<<ver>`` as emitted by
:func:`Version.as_version_spec`, or ``None``/empty for the latest compatible release.
:param for_py_version: major.minor Python version to pass through to ``pip --python-version``.
:param search_dirs: additional directories to treat as a local wheel index when bootstrapping pip.
:param app_data: application data store used to locate the embedded pip wheel.
:param to_folder: directory the downloaded wheel is written into.
:param env: environment mapping passed through to the subprocess.
:returns: the downloaded :class:`Wheel`.
:raises ValueError: if ``distribution`` or ``version_spec`` fail the strict allow-list check.
:raises CalledProcessError: if ``pip download`` exits with a non-zero status.
"""
_check_distribution(distribution)
_check_version_spec(version_spec)
to_download = f"{distribution}{version_spec or ''}"
LOGGER.debug("download wheel %s %s to %s", to_download, for_py_version, to_folder)
cmd = [
sys.executable,
"-m",
"pip",
"download",
"--progress-bar",
"off",
"--disable-pip-version-check",
"--only-binary=:all:",
"--no-deps",
"--python-version",
for_py_version,
"-d",
str(to_folder),
to_download,
]
# pip has no interface in python - must be a new sub-process
env = pip_wheel_env_run(search_dirs, app_data, env)
process = Popen(cmd, env=env, stdout=PIPE, stderr=PIPE, universal_newlines=True, encoding="utf-8")
out, err = process.communicate()
if process.returncode != 0:
kwargs = {"output": out, "stderr": err}
raise CalledProcessError(process.returncode, cmd, **kwargs)
result = _find_downloaded_wheel(distribution, version_spec, for_py_version, to_folder, out)
LOGGER.debug("downloaded wheel %s", result.name) # ty: ignore[unresolved-attribute]
return result # ty: ignore[invalid-return-type]
def _find_downloaded_wheel(
distribution: str, version_spec: str | None, for_py_version: str, to_folder: Path, out: str
) -> Wheel | None:
for line in out.splitlines():
stripped_line = line.lstrip()
for marker in ("Saved ", "File was already downloaded "):
if stripped_line.startswith(marker):
return Wheel(Path(stripped_line[len(marker) :]).absolute())
# if for some reason the output does not match fallback to the latest version with that spec
return find_compatible_in_house(distribution, version_spec, for_py_version, to_folder)
def find_compatible_in_house(
distribution: str, version_spec: str | None, for_py_version: str, in_folder: Path
) -> Wheel | None:
wheels = discover_wheels(in_folder, distribution, None, for_py_version)
start, end = 0, len(wheels)
if version_spec is not None and version_spec:
if version_spec.startswith("<"):
from_pos, op = 1, lt
elif version_spec.startswith("=="):
from_pos, op = 2, eq
else:
raise ValueError(version_spec)
version = Wheel.as_version_tuple(version_spec[from_pos:])
start = next((at for at, w in enumerate(wheels) if op(w.version_tuple, version)), len(wheels))
return None if start == end else wheels[start]
def pip_wheel_env_run(search_dirs: list[Path], app_data: AppData, env: dict[str, str]) -> dict[str, str]:
env = env.copy()
env.update({"PIP_USE_WHEEL": "1", "PIP_USER": "0", "PIP_NO_INPUT": "1", "PYTHONIOENCODING": "utf-8"})
wheel = get_wheel(
distribution="pip",
version=None,
for_py_version=f"{sys.version_info.major}.{sys.version_info.minor}",
search_dirs=search_dirs,
download=False,
app_data=app_data,
do_periodic_update=False,
env=env,
)
if wheel is None:
msg = "could not find the embedded pip"
raise RuntimeError(msg)
env["PYTHONPATH"] = str(wheel.path)
return env
def _check_distribution(distribution: str) -> None:
if not _DISTRIBUTION_RE.fullmatch(distribution):
msg = f"refusing to download wheel for suspicious distribution name: {distribution!r}"
raise ValueError(msg)
def _check_version_spec(version_spec: str | None) -> None:
if not version_spec:
return
if not _VERSION_SPEC_RE.fullmatch(version_spec):
msg = f"refusing to download wheel with suspicious version spec: {version_spec!r}"
raise ValueError(msg)
__all__ = [
"download_wheel",
"get_wheel",
"pip_wheel_env_run",
]
@@ -0,0 +1,65 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from virtualenv.seed.wheels.embed import get_embed_wheel
from .periodic_update import periodic_update
from .util import Version, Wheel, discover_wheels
if TYPE_CHECKING:
from pathlib import Path
from virtualenv.app_data.base import AppData
def from_bundle( # noqa: PLR0913
distribution: str,
version: str | None,
for_py_version: str,
search_dirs: list[Path],
app_data: AppData,
do_periodic_update: bool,
env: dict[str, str],
) -> Wheel | None:
"""Load the bundled wheel to a cache directory."""
of_version = Version.of_version(version)
wheel = load_embed_wheel(app_data, distribution, for_py_version, of_version)
if version != Version.embed:
# 2. check if we have upgraded embed
if app_data.can_update:
per = do_periodic_update
wheel = periodic_update(distribution, of_version, for_py_version, wheel, search_dirs, app_data, per, env)
# 3. acquire from extra search dir
found_wheel = from_dir(distribution, of_version, for_py_version, search_dirs)
if found_wheel is not None and (wheel is None or found_wheel.version_tuple > wheel.version_tuple):
wheel = found_wheel
return wheel
def load_embed_wheel(app_data: AppData, distribution: str, for_py_version: str, version: str | None) -> Wheel | None:
wheel = get_embed_wheel(distribution, for_py_version)
if wheel is not None:
version_match = version == wheel.version
if version is None or version_match:
with app_data.ensure_extracted(wheel.path, lambda: app_data.house) as wheel_path: # ty: ignore[invalid-argument-type]
wheel = Wheel(wheel_path)
else: # if version does not match ignore
wheel = None
return wheel
def from_dir(distribution: str, version: str | None, for_py_version: str, directories: list[Path]) -> Wheel | None:
"""Load a compatible wheel from a given folder."""
for folder in directories:
for wheel in discover_wheels(folder, distribution, version, for_py_version):
return wheel
return None
__all__ = [
"from_bundle",
"load_embed_wheel",
]
@@ -0,0 +1,135 @@
from __future__ import annotations
import hashlib
import zipfile
from pathlib import Path
from virtualenv.info import IS_ZIPAPP, ROOT
from virtualenv.seed.wheels.util import Wheel
BUNDLE_FOLDER = Path(__file__).absolute().parent
BUNDLE_SUPPORT = {
"3.9": {
"pip": "pip-26.0.1-py3-none-any.whl",
"setuptools": "setuptools-82.0.1-py3-none-any.whl",
},
"3.10": {
"pip": "pip-26.1.2-py3-none-any.whl",
"setuptools": "setuptools-83.0.0-py3-none-any.whl",
},
"3.11": {
"pip": "pip-26.1.2-py3-none-any.whl",
"setuptools": "setuptools-83.0.0-py3-none-any.whl",
},
"3.12": {
"pip": "pip-26.1.2-py3-none-any.whl",
"setuptools": "setuptools-83.0.0-py3-none-any.whl",
},
"3.13": {
"pip": "pip-26.1.2-py3-none-any.whl",
"setuptools": "setuptools-83.0.0-py3-none-any.whl",
},
"3.14": {
"pip": "pip-26.1.2-py3-none-any.whl",
"setuptools": "setuptools-83.0.0-py3-none-any.whl",
},
"3.15": {
"pip": "pip-26.1.2-py3-none-any.whl",
"setuptools": "setuptools-83.0.0-py3-none-any.whl",
},
"3.16": {
"pip": "pip-26.1.2-py3-none-any.whl",
"setuptools": "setuptools-83.0.0-py3-none-any.whl",
},
}
MAX = next(reversed(BUNDLE_SUPPORT))
MIN = next(iter(BUNDLE_SUPPORT))
def _release_tuple(version: str) -> tuple[int, ...]:
return tuple(int(part) for part in version.split("."))
# oldest target Python version virtualenv still bundles seed wheels for; anything below this has no embedded pip
OLDEST_SUPPORTED = _release_tuple(MIN)
# SHA-256 of every bundled wheel. Verified on load so a corrupted or tampered wheel on disk fails loud instead of
# being handed to pip. Generated together with ``BUNDLE_SUPPORT`` by ``tasks/upgrade_wheels.py``.
BUNDLE_SHA256 = {
"pip-26.0.1-py3-none-any.whl": "bdb1b08f4274833d62c1aa29e20907365a2ceb950410df15fc9521bad440122b",
"pip-26.1.2-py3-none-any.whl": "382ff9f685ee3bc25864f820aa50505825f10f5458ffff07e30a6d96e5715cab",
"setuptools-82.0.1-py3-none-any.whl": "a59e362652f08dcd477c78bb6e7bd9d80a7995bc73ce773050228a348ce2e5bb",
"setuptools-83.0.0-py3-none-any.whl": "29b23c360f22f414dc7336bb39178cc7bcbf6021ed2733cde173f09dba19abb3",
}
_VERIFIED_WHEELS: set[str] = set()
def get_embed_wheel(distribution: str, for_py_version: str | None) -> Wheel | None:
"""Return the bundled wheel that ships with virtualenv for a given distribution and Python version.
:param distribution: project name of the seed package, for example ``pip`` or ``setuptools``.
:param for_py_version: major.minor Python version string the environment will be created for, or ``None`` to use the
newest bundle.
:returns: a :class:`Wheel` pointing at the verified bundled file, or ``None`` when no wheel is bundled for the
requested combination, including target versions below the oldest bundled one.
:raises RuntimeError: if the bundled wheel on disk fails SHA-256 verification.
"""
if for_py_version is None or _release_tuple(for_py_version) > _release_tuple(MAX):
# no specific target, or a Python newer than anything bundled: reuse the newest bundle
mapping = BUNDLE_SUPPORT[MAX]
else: # versions below the oldest bundled one fall through to None instead of an incompatible newer wheel
mapping = BUNDLE_SUPPORT.get(for_py_version)
if not mapping:
return None
wheel_file = mapping.get(distribution)
if wheel_file is None:
return None
path = BUNDLE_FOLDER / wheel_file
_verify_bundled_wheel(path)
return Wheel.from_path(path)
def _verify_bundled_wheel(path: Path) -> None:
name = path.name
if name in _VERIFIED_WHEELS:
return
expected = BUNDLE_SHA256.get(name)
if expected is None:
msg = f"bundled wheel {name} has no recorded sha256 in BUNDLE_SHA256"
raise RuntimeError(msg)
actual = _hash_bundled_wheel(path)
if actual != expected:
msg = f"bundled wheel {name} sha256 mismatch: expected {expected}, got {actual}"
raise RuntimeError(msg)
_VERIFIED_WHEELS.add(name)
def _hash_bundled_wheel(path: Path) -> str:
# ``path`` is under the package directory; when virtualenv runs from a zipapp the wheel lives inside the
# archive and cannot be opened as a regular file, so read the bytes straight from the zipapp entry.
digest = hashlib.sha256()
if IS_ZIPAPP:
entry = path.resolve().relative_to(Path(ROOT).resolve()).as_posix()
with zipfile.ZipFile(ROOT, "r") as archive, archive.open(entry) as stream:
for chunk in iter(lambda: stream.read(1 << 20), b""):
digest.update(chunk)
else:
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1 << 20), b""):
digest.update(chunk)
return digest.hexdigest()
__all__ = [
"BUNDLE_FOLDER",
"BUNDLE_SHA256",
"BUNDLE_SUPPORT",
"MAX",
"MIN",
"OLDEST_SUPPORTED",
"get_embed_wheel",
]
@@ -0,0 +1,473 @@
"""Periodically update bundled versions."""
from __future__ import annotations
import json
import logging
import os
import ssl
import sys
from datetime import datetime, timedelta, timezone
from itertools import groupby
from pathlib import Path
from shutil import copy2
from subprocess import DEVNULL, Popen
from textwrap import dedent
from threading import Thread
from typing import TYPE_CHECKING
from urllib.error import URLError
from urllib.request import urlopen
from virtualenv.app_data import AppDataDiskFolder
from virtualenv.seed.wheels.embed import BUNDLE_SUPPORT
from virtualenv.seed.wheels.util import Wheel
from virtualenv.util.subprocess import CREATE_NO_WINDOW
if TYPE_CHECKING:
from collections.abc import Generator
from virtualenv.app_data.base import AppData
LOGGER = logging.getLogger(__name__)
GRACE_PERIOD_CI = timedelta(hours=1) # prevent version switch in the middle of a CI run
GRACE_PERIOD_MINOR = timedelta(days=28)
UPDATE_PERIOD = timedelta(days=14)
UPDATE_ABORTED_DELAY = timedelta(hours=1)
def periodic_update( # noqa: PLR0913
distribution: str,
of_version: str | None,
for_py_version: str,
wheel: Wheel | None,
search_dirs: list[Path],
app_data: AppData,
do_periodic_update: bool,
env: dict[str, str],
) -> Wheel | None:
if do_periodic_update:
handle_auto_update(distribution, for_py_version, wheel, search_dirs, app_data, env)
now = datetime.now(tz=timezone.utc)
def _update_wheel(ver: NewVersion) -> Wheel:
updated_wheel = Wheel(app_data.house / ver.filename)
LOGGER.debug("using %supdated wheel %s", "periodically " if updated_wheel else "", updated_wheel)
return updated_wheel
u_log = UpdateLog.from_app_data(app_data, distribution, for_py_version)
if of_version is None:
for _, group in groupby(u_log.versions, key=lambda v: v.wheel.version_tuple[0:2]):
# use only latest patch version per minor, earlier assumed to be buggy
all_patches = list(group)
ignore_grace_period_minor = any(version for version in all_patches if version.use(now))
for version in all_patches:
if wheel is not None and Path(version.filename).name == wheel.name:
return wheel
if version.use(now, ignore_grace_period_minor):
return _update_wheel(version)
else:
for version in u_log.versions:
if version.wheel.version == of_version:
return _update_wheel(version)
return wheel
def handle_auto_update( # noqa: PLR0913
distribution: str,
for_py_version: str,
wheel: Wheel | None,
search_dirs: list[Path],
app_data: AppData,
env: dict[str, str],
) -> None:
embed_update_log = app_data.embed_update_log(distribution, for_py_version)
u_log = UpdateLog.from_dict(embed_update_log.read())
if u_log.needs_update:
u_log.periodic = True
u_log.started = datetime.now(tz=timezone.utc)
embed_update_log.write(u_log.to_dict())
trigger_update(distribution, for_py_version, wheel, search_dirs, app_data, periodic=True, env=env)
def add_wheel_to_update_log(wheel: Wheel, for_py_version: str, app_data: AppData) -> None:
embed_update_log = app_data.embed_update_log(wheel.distribution, for_py_version)
LOGGER.debug("adding %s information to %s", wheel.name, embed_update_log.file) # ty: ignore[unresolved-attribute]
u_log = UpdateLog.from_dict(embed_update_log.read())
if any(version.filename == wheel.name for version in u_log.versions):
LOGGER.warning("%s already present in %s", wheel.name, embed_update_log.file) # ty: ignore[unresolved-attribute]
return
# we don't need a release date for sources other than "periodic"
version = NewVersion(wheel.name, datetime.now(tz=timezone.utc), None, "download")
u_log.versions.append(version) # always write at the end for proper updates
embed_update_log.write(u_log.to_dict())
DATETIME_FMT = "%Y-%m-%dT%H:%M:%S.%fZ"
def dump_datetime(value: datetime | None) -> str | None:
return None if value is None else value.strftime(DATETIME_FMT)
def load_datetime(value: str | None) -> datetime | None:
return None if value is None else datetime.strptime(value, DATETIME_FMT).replace(tzinfo=timezone.utc)
class NewVersion: # noqa: PLW1641
def __init__(self, filename: str, found_date: datetime, release_date: datetime | None, source: str) -> None:
self.filename = filename
self.found_date = found_date
self.release_date = release_date
self.source = source
@classmethod
def from_dict(cls, dictionary: dict[str, str | None]) -> NewVersion:
return cls(
filename=dictionary["filename"], # ty: ignore[invalid-argument-type]
found_date=load_datetime(dictionary["found_date"]), # ty: ignore[invalid-argument-type]
release_date=load_datetime(dictionary["release_date"]),
source=dictionary["source"], # ty: ignore[invalid-argument-type]
)
def to_dict(self) -> dict[str, str | None]:
return {
"filename": self.filename,
"release_date": dump_datetime(self.release_date),
"found_date": dump_datetime(self.found_date),
"source": self.source,
}
def use(self, now: datetime, ignore_grace_period_minor: bool = False, ignore_grace_period_ci: bool = False) -> bool: # noqa: FBT002
if self.source == "manual":
return True
if self.source == "periodic" and (self.found_date < now - GRACE_PERIOD_CI or ignore_grace_period_ci):
if not ignore_grace_period_minor:
compare_from = self.release_date or self.found_date
return now - compare_from >= GRACE_PERIOD_MINOR
return True
return False
def __repr__(self) -> str:
return (
f"{self.__class__.__name__}(filename={self.filename}), found_date={self.found_date}, "
f"release_date={self.release_date}, source={self.source})"
)
def __eq__(self, other: object) -> bool:
return type(self) == type(other) and all( # noqa: E721
getattr(self, k) == getattr(other, k) for k in ["filename", "release_date", "found_date", "source"]
)
def __ne__(self, other: object) -> bool:
return not (self == other)
@property
def wheel(self) -> Wheel:
return Wheel(Path(self.filename))
class UpdateLog:
def __init__(
self, started: datetime | None, completed: datetime | None, versions: list[NewVersion], periodic: bool | None
) -> None:
self.started = started
self.completed = completed
self.versions = versions
self.periodic = periodic
@classmethod
def from_dict(cls, dictionary: dict[str, object] | None) -> UpdateLog:
if dictionary is None:
dictionary = {}
return cls(
load_datetime(dictionary.get("started")), # ty: ignore[invalid-argument-type]
load_datetime(dictionary.get("completed")), # ty: ignore[invalid-argument-type]
[NewVersion.from_dict(v) for v in dictionary.get("versions", [])], # ty: ignore[not-iterable]
dictionary.get("periodic"), # ty: ignore[invalid-argument-type]
)
@classmethod
def from_app_data(cls, app_data: AppData, distribution: str, for_py_version: str) -> UpdateLog:
raw_json = app_data.embed_update_log(distribution, for_py_version).read()
return cls.from_dict(raw_json)
def to_dict(self) -> dict[str, object]:
return {
"started": dump_datetime(self.started),
"completed": dump_datetime(self.completed),
"periodic": self.periodic,
"versions": [r.to_dict() for r in self.versions],
}
@property
def needs_update(self) -> bool:
now = datetime.now(tz=timezone.utc)
if self.completed is None: # never completed
return self._check_start(now)
if now - self.completed <= UPDATE_PERIOD:
return False
return self._check_start(now)
def _check_start(self, now: datetime) -> bool:
return self.started is None or now - self.started > UPDATE_ABORTED_DELAY
def trigger_update( # noqa: PLR0913
distribution: str,
for_py_version: str,
wheel: Wheel | None,
search_dirs: list[Path],
app_data: AppData,
env: dict[str, str],
periodic: bool,
) -> None:
wheel_path = None if wheel is None else str(wheel.path)
cmd = [
sys.executable,
"-c",
dedent(
"""
from virtualenv.report import setup_report, MAX_LEVEL
from virtualenv.seed.wheels.periodic_update import do_update
setup_report(MAX_LEVEL, show_pid=True)
do_update({!r}, {!r}, {!r}, {!r}, {!r}, {!r})
""",
)
.strip()
.format(distribution, for_py_version, wheel_path, str(app_data), [str(p) for p in search_dirs], periodic),
]
debug = env.get("_VIRTUALENV_PERIODIC_UPDATE_INLINE") == "1"
pipe = None if debug else DEVNULL
kwargs = {"stdout": pipe, "stderr": pipe}
if not debug and sys.platform == "win32":
kwargs["creationflags"] = CREATE_NO_WINDOW
process = Popen(cmd, **kwargs) # ty: ignore[no-matching-overload]
LOGGER.info(
"triggered periodic upgrade of %s%s (for python %s) via background process having PID %d",
distribution,
"" if wheel is None else f"=={wheel.version}",
for_py_version,
process.pid,
)
if debug:
process.communicate() # on purpose not called to make it a background process
else:
# set the returncode here -> no ResourceWarning on main process exit if the subprocess still runs
process.returncode = 0
def do_update( # noqa: PLR0913
distribution: str,
for_py_version: str,
embed_filename: str | None,
app_data: str | AppData,
search_dirs: list[str] | list[Path],
periodic: bool,
) -> list[NewVersion] | None:
versions = None
try:
versions = _run_do_update(app_data, distribution, embed_filename, for_py_version, periodic, search_dirs)
finally:
LOGGER.debug("done %s %s with %s", distribution, for_py_version, versions)
return versions
def _run_do_update( # noqa: C901, PLR0913
app_data: str | AppData,
distribution: str,
embed_filename: str | None,
for_py_version: str,
periodic: bool,
search_dirs: list[str] | list[Path],
) -> list[NewVersion]:
from virtualenv.seed.wheels import acquire # noqa: PLC0415
wheel_filename = None if embed_filename is None else Path(embed_filename)
embed_version = None if wheel_filename is None else Wheel(wheel_filename).version_tuple
app_data = AppDataDiskFolder(app_data) if isinstance(app_data, str) else app_data
search_dirs = [Path(p) if isinstance(p, str) else p for p in search_dirs]
wheelhouse = app_data.house
embed_update_log = app_data.embed_update_log(distribution, for_py_version)
u_log = UpdateLog.from_dict(embed_update_log.read())
now = datetime.now(tz=timezone.utc)
update_versions, other_versions = [], []
for version in u_log.versions:
if version.source in {"periodic", "manual"}:
update_versions.append(version)
else:
other_versions.append(version)
if periodic:
source = "periodic"
else:
source = "manual"
# mark the most recent one as source "manual"
if update_versions:
update_versions[0].source = source
if wheel_filename is not None:
dest = wheelhouse / wheel_filename.name
if not dest.exists():
copy2(str(wheel_filename), str(wheelhouse))
last, last_version, versions, filenames = None, None, [], set()
while last is None or not last.use(now, ignore_grace_period_ci=True):
download_time = datetime.now(tz=timezone.utc)
dest = acquire.download_wheel(
distribution=distribution,
version_spec=None if last_version is None else f"<{last_version}",
for_py_version=for_py_version,
search_dirs=search_dirs,
app_data=app_data,
to_folder=wheelhouse,
env=os.environ, # ty: ignore[invalid-argument-type]
)
if dest is None or (update_versions and update_versions[0].filename == dest.name):
break
release_date = release_date_for_wheel_path(dest.path)
last = NewVersion(filename=dest.path.name, release_date=release_date, found_date=download_time, source=source)
LOGGER.info("detected %s in %s", last, datetime.now(tz=timezone.utc) - download_time)
versions.append(last)
filenames.add(last.filename)
last_wheel = last.wheel
last_version = last_wheel.version
if embed_version is not None and embed_version >= last_wheel.version_tuple:
break # stop download if we reach the embed version
u_log.periodic = periodic
if not u_log.periodic:
u_log.started = now
# update other_versions by removing version we just found
other_versions = [version for version in other_versions if version.filename not in filenames]
u_log.versions = versions + update_versions + other_versions
u_log.completed = datetime.now(tz=timezone.utc)
embed_update_log.write(u_log.to_dict())
return versions
def release_date_for_wheel_path(dest: Path) -> datetime | None:
wheel = Wheel(dest)
# the most accurate is to ask PyPi - e.g. https://pypi.org/pypi/pip/json,
# see https://warehouse.pypa.io/api-reference/json/ for more details
content = _pypi_get_distribution_info_cached(wheel.distribution)
if content is not None:
try:
upload_time = content["releases"][wheel.version][0]["upload_time"] # ty: ignore[not-subscriptable]
return datetime.strptime(upload_time, "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc)
except Exception as exception: # noqa: BLE001
LOGGER.error("could not load release date %s because %r", content, exception) # noqa: TRY400
return None
#: Opt-in escape hatch to restore the pre-2026 behavior of falling back to an unverified HTTPS context when the
#: verified request fails. Off by default: a failed TLS handshake on the PyPI metadata lookup now aborts the update
#: instead of silently downgrading, because the response drives which wheel version virtualenv thinks is up to date.
_INSECURE_FALLBACK_ENV = "VIRTUALENV_PERIODIC_UPDATE_INSECURE"
def _request_context() -> Generator[ssl.SSLContext | None, None, None]:
yield None
if os.environ.get(_INSECURE_FALLBACK_ENV):
LOGGER.warning(
"falling back to unverified HTTPS for PyPI metadata because %s is set",
_INSECURE_FALLBACK_ENV,
)
yield ssl._create_unverified_context() # noqa: S323, SLF001
_PYPI_CACHE = {}
def _pypi_get_distribution_info_cached(distribution: str) -> dict[str, object] | None:
if distribution not in _PYPI_CACHE:
_PYPI_CACHE[distribution] = _pypi_get_distribution_info(distribution)
return _PYPI_CACHE[distribution]
def _pypi_get_distribution_info(distribution: str) -> dict[str, object] | None:
content, url = None, f"https://pypi.org/pypi/{distribution}/json"
try:
for context in _request_context():
if (content := _fetch_pypi_json(url, context)) is not None:
break
except Exception as exception: # noqa: BLE001
LOGGER.error("failed to access %s because %r", url, exception) # noqa: TRY400
return content
def _fetch_pypi_json(url: str, context: ssl.SSLContext | None) -> dict[str, object] | None:
try:
with urlopen(url, context=context) as file_handler: # noqa: S310
return json.load(file_handler)
except URLError as exception:
LOGGER.error("failed to access %s because %r", url, exception) # noqa: TRY400
return None
def manual_upgrade(app_data: AppData, env: dict[str, str]) -> None:
threads = []
for for_py_version, distribution_to_package in BUNDLE_SUPPORT.items():
# load extra search dir for the given for_py
for distribution in distribution_to_package:
thread = Thread(target=_run_manual_upgrade, args=(app_data, distribution, for_py_version, env))
thread.start()
threads.append(thread)
for thread in threads:
thread.join()
def _run_manual_upgrade(app_data: AppData, distribution: str, for_py_version: str, env: dict[str, str]) -> None:
start = datetime.now(tz=timezone.utc)
from .bundle import from_bundle # noqa: PLC0415
current = from_bundle(
distribution=distribution,
version=None,
for_py_version=for_py_version,
search_dirs=[],
app_data=app_data,
do_periodic_update=False,
env=env,
)
LOGGER.warning(
"upgrade %s for python %s with current %s",
distribution,
for_py_version,
"" if current is None else current.name,
)
versions = do_update(
distribution=distribution,
for_py_version=for_py_version,
embed_filename=current.path, # ty: ignore[invalid-argument-type, unresolved-attribute]
app_data=app_data,
search_dirs=[],
periodic=False,
)
args = [
distribution,
for_py_version,
datetime.now(tz=timezone.utc) - start,
]
if versions:
args.append("\n".join(f"\t{v}" for v in versions))
ver_update = "new entries found:\n%s" if versions else "no new versions found"
msg = f"upgraded %s for python %s in %s {ver_update}"
LOGGER.warning(msg, *args)
__all__ = [
"NewVersion",
"UpdateLog",
"add_wheel_to_update_log",
"do_update",
"dump_datetime",
"load_datetime",
"manual_upgrade",
"periodic_update",
"release_date_for_wheel_path",
"trigger_update",
]
@@ -0,0 +1,125 @@
from __future__ import annotations
from operator import attrgetter
from typing import TYPE_CHECKING
from zipfile import ZipFile
if TYPE_CHECKING:
from pathlib import Path
class Wheel:
def __init__(self, path: Path) -> None:
# https://www.python.org/dev/peps/pep-0427/#file-name-convention
# The wheel filename is {distribution}-{version}(-{build tag})?-{python tag}-{abi tag}-{platform tag}.whl
self.path = path
self._parts = path.stem.split("-")
@classmethod
def from_path(cls, path: Path) -> Wheel | None:
if path is not None and path.suffix == ".whl" and len(path.stem.split("-")) >= 5: # noqa: PLR2004
return cls(path)
return None
@property
def distribution(self) -> str:
return self._parts[0]
@property
def version(self) -> str:
return self._parts[1]
@property
def version_tuple(self) -> tuple[int, ...]:
return self.as_version_tuple(self.version)
@staticmethod
def as_version_tuple(version: str) -> tuple[int, ...]:
result = []
for part in version.split(".")[0:3]:
try:
result.append(int(part))
except ValueError: # noqa: PERF203
break
if not result:
raise ValueError(version)
return tuple(result)
@property
def name(self) -> str:
return self.path.name
def support_py(self, py_version: str) -> bool:
name = f"{'-'.join(self.path.stem.split('-')[0:2])}.dist-info/METADATA"
with ZipFile(str(self.path), "r") as zip_file:
metadata = zip_file.read(name).decode("utf-8")
marker = "Requires-Python:"
requires = next((i[len(marker) :] for i in metadata.splitlines() if i.startswith(marker)), None)
if requires is None: # if it does not specify a python requires the assumption is compatible
return True
py_version_int = tuple(int(i) for i in py_version.split("."))
for require in (i.strip() for i in requires.split(",")):
# https://www.python.org/dev/peps/pep-0345/#version-specifiers
for operator, check in [
("!=", lambda v: py_version_int != v),
("==", lambda v: py_version_int == v),
("<=", lambda v: py_version_int <= v),
(">=", lambda v: py_version_int >= v),
("<", lambda v: py_version_int < v),
(">", lambda v: py_version_int > v),
]:
if require.startswith(operator):
ver_str = require[len(operator) :].strip()
version = tuple((int(i) if i != "*" else None) for i in ver_str.split("."))[0:2]
if not check(version):
return False
break
return True
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self.path})"
def __str__(self) -> str:
return str(self.path)
def discover_wheels(from_folder: Path, distribution: str, version: str | None, for_py_version: str) -> list[Wheel]:
wheels = []
for filename in from_folder.iterdir():
wheel = Wheel.from_path(filename)
if (
wheel
and wheel.distribution == distribution
and (version is None or wheel.version == version)
and wheel.support_py(for_py_version)
):
wheels.append(wheel)
return sorted(wheels, key=attrgetter("version_tuple", "distribution"), reverse=True)
class Version:
#: the version bundled with virtualenv
bundle = "bundle"
embed = "embed"
#: custom version handlers
non_version = (bundle, embed)
@staticmethod
def of_version(value: str | None) -> str | None:
return None if value in Version.non_version else value
@staticmethod
def as_pip_req(distribution: str, version: str | None) -> str:
return f"{distribution}{Version.as_version_spec(version)}"
@staticmethod
def as_version_spec(version: str | None) -> str:
of_version = Version.of_version(version)
return "" if of_version is None else f"=={of_version}"
__all__ = [
"Version",
"Wheel",
"discover_wheels",
]