File: /home/imagzxcb/public_html/db127a/create.zip
PK 鼷\~�W< <
creator.pynu �[��� import json
import logging
import os
import sys
from abc import ABCMeta, abstractmethod
from argparse import ArgumentTypeError
from ast import literal_eval
from collections import OrderedDict
from pathlib import Path
from virtualenv.discovery.cached_py_info import LogCmd
from virtualenv.util.path import safe_delete
from virtualenv.util.subprocess import run_cmd
from virtualenv.version import __version__
from .pyenv_cfg import PyEnvCfg
HERE = Path(os.path.abspath(__file__)).parent
DEBUG_SCRIPT = HERE / "debug.py"
class CreatorMeta:
def __init__(self):
self.error = None
class Creator(metaclass=ABCMeta):
"""A class that given a python Interpreter creates a virtual environment"""
def __init__(self, options, interpreter):
"""Construct a new virtual environment creator.
:param options: the CLI option as parsed from :meth:`add_parser_arguments`
:param interpreter: the interpreter to create virtual environment from
"""
self.interpreter = interpreter
self._debug = None
self.dest = Path(options.dest)
self.clear = options.clear
self.no_vcs_ignore = options.no_vcs_ignore
self.pyenv_cfg = PyEnvCfg.from_folder(self.dest)
self.app_data = options.app_data
self.env = options.env
def __repr__(self):
return f"{self.__class__.__name__}({', '.join(f'{k}={v}' for k, v in self._args())})"
def _args(self):
return [
("dest", str(self.dest)),
("clear", self.clear),
("no_vcs_ignore", self.no_vcs_ignore),
]
@classmethod
def can_create(cls, interpreter): # noqa: U100
"""Determine if we can create a virtual environment.
:param interpreter: the interpreter in question
:return: ``None`` if we can't create, any other object otherwise that will be forwarded to \
:meth:`add_parser_arguments`
"""
return True
@classmethod
def add_parser_arguments(cls, parser, interpreter, meta, app_data): # noqa: U100
"""Add CLI arguments for the creator.
:param parser: the CLI parser
:param app_data: the application data folder
:param interpreter: the interpreter we're asked to create virtual environment for
:param meta: value as returned by :meth:`can_create`
"""
parser.add_argument(
"dest",
help="directory to create virtualenv at",
type=cls.validate_dest,
)
parser.add_argument(
"--clear",
dest="clear",
action="store_true",
help="remove the destination directory if exist before starting (will overwrite files otherwise)",
default=False,
)
parser.add_argument(
"--no-vcs-ignore",
dest="no_vcs_ignore",
action="store_true",
help="don't create VCS ignore directive in the destination directory",
default=False,
)
@abstractmethod
def create(self):
"""Perform the virtual environment creation."""
raise NotImplementedError
@classmethod
def validate_dest(cls, raw_value):
"""No path separator in the path, valid chars and must be write-able"""
def non_write_able(dest, value):
common = Path(*os.path.commonprefix([value.parts, dest.parts]))
raise ArgumentTypeError(f"the destination {dest.relative_to(common)} is not write-able at {common}")
# the file system must be able to encode
# note in newer CPython this is always utf-8 https://www.python.org/dev/peps/pep-0529/
encoding = sys.getfilesystemencoding()
refused = OrderedDict()
kwargs = {"errors": "ignore"} if encoding != "mbcs" else {}
for char in str(raw_value):
try:
trip = char.encode(encoding, **kwargs).decode(encoding)
if trip == char:
continue
raise ValueError(trip)
except ValueError:
refused[char] = None
if refused:
bad = "".join(refused.keys())
msg = f"the file system codec ({encoding}) cannot handle characters {bad!r} within {raw_value!r}"
raise ArgumentTypeError(msg)
if os.pathsep in raw_value:
msg = f"destination {raw_value!r} must not contain the path separator ({os.pathsep})"
raise ArgumentTypeError(f"{msg} as this would break the activation scripts")
value = Path(raw_value)
if value.exists() and value.is_file():
raise ArgumentTypeError(f"the destination {value} already exists and is a file")
dest = Path(os.path.abspath(str(value))).resolve() # on Windows absolute does not imply resolve so use both
value = dest
while dest:
if dest.exists():
if os.access(str(dest), os.W_OK):
break
else:
non_write_able(dest, value)
base, _ = dest.parent, dest.name
if base == dest:
non_write_able(dest, value) # pragma: no cover
dest = base
return str(value)
def run(self):
if self.dest.exists() and self.clear:
logging.debug("delete %s", self.dest)
safe_delete(self.dest)
self.create()
self.set_pyenv_cfg()
if not self.no_vcs_ignore:
self.setup_ignore_vcs()
def set_pyenv_cfg(self):
self.pyenv_cfg.content = OrderedDict()
self.pyenv_cfg["home"] = os.path.dirname(os.path.abspath(self.interpreter.system_executable))
self.pyenv_cfg["implementation"] = self.interpreter.implementation
self.pyenv_cfg["version_info"] = ".".join(str(i) for i in self.interpreter.version_info)
self.pyenv_cfg["virtualenv"] = __version__
def setup_ignore_vcs(self):
"""Generate ignore instructions for version control systems."""
# mark this folder to be ignored by VCS, handle https://www.python.org/dev/peps/pep-0610/#registered-vcs
git_ignore = self.dest / ".gitignore"
if not git_ignore.exists():
git_ignore.write_text("# created by virtualenv automatically\n*\n", encoding="utf-8")
# Mercurial - does not support the .hgignore file inside a subdirectory directly, but only if included via the
# subinclude directive from root, at which point on might as well ignore the directory itself, see
# https://www.selenic.com/mercurial/hgignore.5.html for more details
# Bazaar - does not support ignore files in sub-directories, only at root level via .bzrignore
# Subversion - does not support ignore files, requires direct manipulation with the svn tool
@property
def debug(self):
"""
:return: debug information about the virtual environment (only valid after :meth:`create` has run)
"""
if self._debug is None and self.exe is not None:
self._debug = get_env_debug_info(self.exe, self.debug_script(), self.app_data, self.env)
return self._debug
@staticmethod
def debug_script():
return DEBUG_SCRIPT
def get_env_debug_info(env_exe, debug_script, app_data, env):
env = env.copy()
env.pop("PYTHONPATH", None)
with app_data.ensure_extracted(debug_script) as debug_script:
cmd = [str(env_exe), str(debug_script)]
logging.debug("debug via %r", LogCmd(cmd))
code, out, err = run_cmd(cmd)
try:
if code != 0:
if out:
result = literal_eval(out)
else:
if code == 2 and "file" in err:
# Re-raise FileNotFoundError from `run_cmd()`
raise OSError(err)
raise Exception(err)
else:
result = json.loads(out)
if err:
result["err"] = err
except Exception as exception:
return {"out": out, "err": err, "returncode": code, "exception": repr(exception)}
if "sys" in result and "path" in result["sys"]:
del result["sys"]["path"][0]
return result
__all__ = [
"Creator",
"CreatorMeta",
]
PK 鼷\����[ [ via_global_ref/_virtualenv.pynu �[��� """Patches that are applied at runtime to the virtual environment"""
# -*- coding: utf-8 -*-
import os
import sys
VIRTUALENV_PATCH_FILE = os.path.join(__file__)
def patch_dist(dist):
"""
Distutils allows user to configure some arguments via a configuration file:
https://docs.python.org/3/install/index.html#distutils-configuration-files
Some of this arguments though don't make sense in context of the virtual environment files, let's fix them up.
"""
# we cannot allow some install config as that would get packages installed outside of the virtual environment
old_parse_config_files = dist.Distribution.parse_config_files
def parse_config_files(self, *args, **kwargs):
result = old_parse_config_files(self, *args, **kwargs)
install = self.get_option_dict("install")
if "prefix" in install: # the prefix governs where to install the libraries
install["prefix"] = VIRTUALENV_PATCH_FILE, os.path.abspath(sys.prefix)
for base in ("purelib", "platlib", "headers", "scripts", "data"):
key = "install_{}".format(base)
if key in install: # do not allow global configs to hijack venv paths
install.pop(key, None)
return result
dist.Distribution.parse_config_files = parse_config_files
# Import hook that patches some modules to ignore configuration values that break package installation in case
# of virtual environments.
_DISTUTILS_PATCH = "distutils.dist", "setuptools.dist"
if sys.version_info > (3, 4):
# https://docs.python.org/3/library/importlib.html#setting-up-an-importer
class _Finder:
"""A meta path finder that allows patching the imported distutils modules"""
fullname = None
# lock[0] is threading.Lock(), but initialized lazily to avoid importing threading very early at startup,
# because there are gevent-based applications that need to be first to import threading by themselves.
# See https://github.com/pypa/virtualenv/issues/1895 for details.
lock = []
def find_spec(self, fullname, path, target=None): # noqa: U100
if fullname in _DISTUTILS_PATCH and self.fullname is None:
# initialize lock[0] lazily
if len(self.lock) == 0:
import threading
lock = threading.Lock()
# there is possibility that two threads T1 and T2 are simultaneously running into find_spec,
# observing .lock as empty, and further going into hereby initialization. However due to the GIL,
# list.append() operation is atomic and this way only one of the threads will "win" to put the lock
# - that every thread will use - into .lock[0].
# https://docs.python.org/3/faq/library.html#what-kinds-of-global-value-mutation-are-thread-safe
self.lock.append(lock)
from functools import partial
from importlib.util import find_spec
with self.lock[0]:
self.fullname = fullname
try:
spec = find_spec(fullname, path)
if spec is not None:
# https://www.python.org/dev/peps/pep-0451/#how-loading-will-work
is_new_api = hasattr(spec.loader, "exec_module")
func_name = "exec_module" if is_new_api else "load_module"
old = getattr(spec.loader, func_name)
func = self.exec_module if is_new_api else self.load_module
if old is not func:
try:
setattr(spec.loader, func_name, partial(func, old))
except AttributeError:
pass # C-Extension loaders are r/o such as zipimporter with <python 3.7
return spec
finally:
self.fullname = None
@staticmethod
def exec_module(old, module):
old(module)
if module.__name__ in _DISTUTILS_PATCH:
patch_dist(module)
@staticmethod
def load_module(old, name):
module = old(name)
if module.__name__ in _DISTUTILS_PATCH:
patch_dist(module)
return module
sys.meta_path.insert(0, _Finder())
elif sys.version_info < (3, 3):
# https://www.python.org/dev/peps/pep-0302/
from imp import find_module
from pkgutil import ImpImporter, ImpLoader
class _VirtualenvImporter(object, ImpImporter):
def __init__(self, path=None):
object.__init__(self)
ImpImporter.__init__(self, path)
def find_module(self, fullname, path=None):
if fullname in _DISTUTILS_PATCH:
try:
return _VirtualenvLoader(fullname, *find_module(fullname.split(".")[-1], path))
except ImportError:
pass
return None
class _VirtualenvLoader(object, ImpLoader):
def __init__(self, fullname, file, filename, etc):
object.__init__(self)
ImpLoader.__init__(self, fullname, file, filename, etc)
def load_module(self, fullname):
module = super(_VirtualenvLoader, self).load_module(fullname)
patch_dist(module)
module.__loader__ = None # distlib fallback
return module
sys.meta_path.append(_VirtualenvImporter())
else:
# https://www.python.org/dev/peps/pep-0302/
from imp import find_module
from pkgutil import ImpImporter, ImpLoader
class _VirtualenvImporter(ImpImporter):
def __init__(self, path=None):
object.__init__(self)
ImpImporter.__init__(self, path)
def find_module(self, fullname, path=None):
if fullname in _DISTUTILS_PATCH:
try:
return _VirtualenvLoader(fullname, *find_module(fullname.split(".")[-1], path))
except ImportError:
pass
return None
class _VirtualenvLoader(ImpLoader):
def __init__(self, fullname, file, filename, etc):
object.__init__(self)
ImpLoader.__init__(self, fullname, file, filename, etc)
def load_module(self, fullname):
module = super(_VirtualenvLoader, self).load_module(fullname)
patch_dist(module)
module.__loader__ = None # distlib fallback
return module
sys.meta_path.append(_VirtualenvImporter())
PK 鼷\XC�f f via_global_ref/store.pynu �[��� from pathlib import Path
def handle_store_python(meta, interpreter):
if is_store_python(interpreter):
meta.symlink_error = "Windows Store Python does not support virtual environments via symlink"
return meta
def is_store_python(interpreter):
parts = Path(interpreter.system_executable).parts
return (
len(parts) > 4
and parts[-4] == "Microsoft"
and parts[-3] == "WindowsApps"
and parts[-2].startswith("PythonSoftwareFoundation.Python.3.")
and parts[-1].startswith("python")
)
__all__ = [
"handle_store_python",
"is_store_python",
]
PK 鼷\k4��T T via_global_ref/api.pynu �[��� import logging
import os
from abc import ABCMeta
from pathlib import Path
from virtualenv.info import fs_supports_symlink
from ..creator import Creator, CreatorMeta
class ViaGlobalRefMeta(CreatorMeta):
def __init__(self):
super().__init__()
self.copy_error = None
self.symlink_error = None
if not fs_supports_symlink():
self.symlink_error = "the filesystem does not supports symlink"
@property
def can_copy(self):
return not self.copy_error
@property
def can_symlink(self):
return not self.symlink_error
class ViaGlobalRefApi(Creator, metaclass=ABCMeta):
def __init__(self, options, interpreter):
super().__init__(options, interpreter)
self.symlinks = self._should_symlink(options)
self.enable_system_site_package = options.system_site
@staticmethod
def _should_symlink(options):
# Priority of where the option is set to follow the order: CLI, env var, file, hardcoded.
# If both set at same level prefers copy over symlink.
copies, symlinks = getattr(options, "copies", False), getattr(options, "symlinks", False)
copy_src, sym_src = options.get_source("copies"), options.get_source("symlinks")
for level in ["cli", "env var", "file", "default"]:
s_opt = symlinks if sym_src == level else None
c_opt = copies if copy_src == level else None
if s_opt is True and c_opt is True:
return False
if s_opt is True:
return True
if c_opt is True:
return False
return False # fallback to copy
@classmethod
def add_parser_arguments(cls, parser, interpreter, meta, app_data):
super().add_parser_arguments(parser, interpreter, meta, app_data)
parser.add_argument(
"--system-site-packages",
default=False,
action="store_true",
dest="system_site",
help="give the virtual environment access to the system site-packages dir",
)
group = parser.add_mutually_exclusive_group()
if not meta.can_symlink and not meta.can_copy:
raise RuntimeError("neither symlink or copy method supported")
if meta.can_symlink:
group.add_argument(
"--symlinks",
default=True,
action="store_true",
dest="symlinks",
help="try to use symlinks rather than copies, when symlinks are not the default for the platform",
)
if meta.can_copy:
group.add_argument(
"--copies",
"--always-copy",
default=not meta.can_symlink,
action="store_true",
dest="copies",
help="try to use copies rather than symlinks, even when symlinks are the default for the platform",
)
def create(self):
self.install_patch()
def install_patch(self):
text = self.env_patch_text()
if text:
pth = self.purelib / "_virtualenv.pth"
logging.debug("create virtualenv import hook file %s", pth)
pth.write_text("import _virtualenv", encoding="utf-8")
dest_path = self.purelib / "_virtualenv.py"
logging.debug("create %s", dest_path)
dest_path.write_text(text, encoding="utf-8")
def env_patch_text(self):
"""Patch the distutils package to not be derailed by its configuration files"""
with self.app_data.ensure_extracted(Path(__file__).parent / "_virtualenv.py") as resolved_path:
text = resolved_path.read_text(encoding="utf-8")
return text.replace('"__SCRIPT_DIR__"', repr(os.path.relpath(str(self.script_dir), str(self.purelib))))
def _args(self):
return super()._args() + [("global", self.enable_system_site_package)]
def set_pyenv_cfg(self):
super().set_pyenv_cfg()
self.pyenv_cfg["include-system-site-packages"] = "true" if self.enable_system_site_package else "false"
__all__ = [
"ViaGlobalRefMeta",
"ViaGlobalRefApi",
]
PK 鼷\V��� � &