codekingpro/portable-devtools

1"""Patches that are applied at runtime to the virtual environment."""
2
3from __future__ import annotations
4
5import os
6import sys
7
8VIRTUALENV_PATCH_FILE = os.path.join(__file__)
9
10
11def patch_dist(dist):
12    """
13    Distutils allows user to configure some arguments via a configuration file:
14    https://docs.python.org/3/install/index.html#distutils-configuration-files.
15
16    Some of this arguments though don't make sense in context of the virtual environment files, let's fix them up.
17    """  # noqa: D205
18    # we cannot allow some install config as that would get packages installed outside of the virtual environment
19    old_parse_config_files = dist.Distribution.parse_config_files
20
21    def parse_config_files(self, *args, **kwargs):
22        result = old_parse_config_files(self, *args, **kwargs)
23        install = self.get_option_dict("install")
24
25        if "prefix" in install:  # the prefix governs where to install the libraries
26            install["prefix"] = VIRTUALENV_PATCH_FILE, os.path.abspath(sys.prefix)
27        for base in ("purelib", "platlib", "headers", "scripts", "data"):
28            key = f"install_{base}"
29            if key in install:  # do not allow global configs to hijack venv paths
30                install.pop(key, None)
31        return result
32
33    dist.Distribution.parse_config_files = parse_config_files
34
35
36# Import hook that patches some modules to ignore configuration values that break package installation in case
37# of virtual environments.
38_DISTUTILS_PATCH = "distutils.dist", "setuptools.dist"
39# https://docs.python.org/3/library/importlib.html#setting-up-an-importer
40
41
42class _Finder:
43    """A meta path finder that allows patching the imported distutils modules."""
44
45    fullname = None
46
47    # lock[0] is threading.Lock(), but initialized lazily to avoid importing threading very early at startup,
48    # because there are gevent-based applications that need to be first to import threading by themselves.
49    # See https://github.com/pypa/virtualenv/issues/1895 for details.
50    lock = []  # noqa: RUF012
51
52    def find_spec(self, fullname, path, target=None):  # noqa: ARG002
53        if fullname in _DISTUTILS_PATCH and self.fullname is None:  # noqa: PLR1702
54            # initialize lock[0] lazily
55            if len(self.lock) == 0:
56                import threading  # noqa: PLC0415
57
58                lock = threading.Lock()
59                # there is possibility that two threads T1 and T2 are simultaneously running into find_spec,
60                # observing .lock as empty, and further going into hereby initialization. However due to the GIL,
61                # list.append() operation is atomic and this way only one of the threads will "win" to put the lock
62                # - that every thread will use - into .lock[0].
63                # https://docs.python.org/3/faq/library.html#what-kinds-of-global-value-mutation-are-thread-safe
64                self.lock.append(lock)
65
66            from functools import partial  # noqa: PLC0415
67            from importlib.util import find_spec  # noqa: PLC0415
68
69            with self.lock[0]:
70                self.fullname = fullname
71                try:
72                    spec = find_spec(fullname, path)
73                    if spec is not None:
74                        # https://www.python.org/dev/peps/pep-0451/#how-loading-will-work
75                        is_new_api = hasattr(spec.loader, "exec_module")
76                        func_name = "exec_module" if is_new_api else "load_module"
77                        old = getattr(spec.loader, func_name)
78                        func = self.exec_module if is_new_api else self.load_module
79                        if old is not func:
80                            try:  # noqa: SIM105
81                                setattr(spec.loader, func_name, partial(func, old))
82                            except AttributeError:
83                                pass  # C-Extension loaders are r/o such as zipimporter with <3.7
84                        return spec
85                finally:
86                    self.fullname = None
87        return None
88
89    @staticmethod
90    def exec_module(old, module):
91        old(module)
92        if module.__name__ in _DISTUTILS_PATCH:
93            patch_dist(module)
94
95    @staticmethod
96    def load_module(old, name):
97        module = old(name)
98        if module.__name__ in _DISTUTILS_PATCH:
99            patch_dist(module)
100        return module
101
102
103sys.meta_path.insert(0, _Finder())
104