diff planemo/lib/python3.7/site-packages/virtualenv/discovery/cached_py_info.py @ 1:56ad4e20f292 draft

"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:32:28 -0400
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/planemo/lib/python3.7/site-packages/virtualenv/discovery/cached_py_info.py	Fri Jul 31 00:32:28 2020 -0400
@@ -0,0 +1,145 @@
+"""
+
+We acquire the python information by running an interrogation script via subprocess trigger. This operation is not
+cheap, especially not on Windows. To not have to pay this hefty cost every time we apply multiple levels of
+caching.
+"""
+from __future__ import absolute_import, unicode_literals
+
+import logging
+import os
+import pipes
+import sys
+from collections import OrderedDict
+
+from virtualenv.app_data import AppDataDisabled
+from virtualenv.discovery.py_info import PythonInfo
+from virtualenv.info import PY2
+from virtualenv.util.path import Path
+from virtualenv.util.six import ensure_text
+from virtualenv.util.subprocess import Popen, subprocess
+
+_CACHE = OrderedDict()
+_CACHE[Path(sys.executable)] = PythonInfo()
+
+
+def from_exe(cls, app_data, exe, raise_on_error=True, ignore_cache=False):
+    """"""
+    result = _get_from_cache(cls, app_data, exe, ignore_cache=ignore_cache)
+    if isinstance(result, Exception):
+        if raise_on_error:
+            raise result
+        else:
+            logging.info("%s", str(result))
+        result = None
+    return result
+
+
+def _get_from_cache(cls, app_data, exe, ignore_cache=True):
+    # note here we cannot resolve symlinks, as the symlink may trigger different prefix information if there's a
+    # pyenv.cfg somewhere alongside on python3.4+
+    exe_path = Path(exe)
+    if not ignore_cache and exe_path in _CACHE:  # check in the in-memory cache
+        result = _CACHE[exe_path]
+    else:  # otherwise go through the app data cache
+        py_info = _get_via_file_cache(cls, app_data, exe_path, exe)
+        result = _CACHE[exe_path] = py_info
+    # independent if it was from the file or in-memory cache fix the original executable location
+    if isinstance(result, PythonInfo):
+        result.executable = exe
+    return result
+
+
+def _get_via_file_cache(cls, app_data, path, exe):
+    path_text = ensure_text(str(path))
+    try:
+        path_modified = path.stat().st_mtime
+    except OSError:
+        path_modified = -1
+    if app_data is None:
+        app_data = AppDataDisabled()
+    py_info, py_info_store = None, app_data.py_info(path)
+    with py_info_store.locked():
+        if py_info_store.exists():  # if exists and matches load
+            data = py_info_store.read()
+            of_path, of_st_mtime, of_content = data["path"], data["st_mtime"], data["content"]
+            if of_path == path_text and of_st_mtime == path_modified:
+                py_info = cls._from_dict({k: v for k, v in of_content.items()})
+            else:
+                py_info_store.remove()
+        if py_info is None:  # if not loaded run and save
+            failure, py_info = _run_subprocess(cls, exe, app_data)
+            if failure is None:
+                data = {"st_mtime": path_modified, "path": path_text, "content": py_info._to_dict()}
+                py_info_store.write(data)
+            else:
+                py_info = failure
+    return py_info
+
+
+def _run_subprocess(cls, exe, app_data):
+    py_info_script = Path(os.path.abspath(__file__)).parent / "py_info.py"
+    with app_data.ensure_extracted(py_info_script) as py_info_script:
+        cmd = [exe, str(py_info_script)]
+        # prevent sys.prefix from leaking into the child process - see https://bugs.python.org/issue22490
+        env = os.environ.copy()
+        env.pop("__PYVENV_LAUNCHER__", None)
+        logging.debug("get interpreter info via cmd: %s", LogCmd(cmd))
+        try:
+            process = Popen(
+                cmd,
+                universal_newlines=True,
+                stdin=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                env=env,
+            )
+            out, err = process.communicate()
+            code = process.returncode
+        except OSError as os_error:
+            out, err, code = "", os_error.strerror, os_error.errno
+    result, failure = None, None
+    if code == 0:
+        result = cls._from_json(out)
+        result.executable = exe  # keep original executable as this may contain initialization code
+    else:
+        msg = "failed to query {} with code {}{}{}".format(
+            exe, code, " out: {!r}".format(out) if out else "", " err: {!r}".format(err) if err else "",
+        )
+        failure = RuntimeError(msg)
+    return failure, result
+
+
+class LogCmd(object):
+    def __init__(self, cmd, env=None):
+        self.cmd = cmd
+        self.env = env
+
+    def __repr__(self):
+        def e(v):
+            return v.decode("utf-8") if isinstance(v, bytes) else v
+
+        cmd_repr = e(" ").join(pipes.quote(e(c)) for c in self.cmd)
+        if self.env is not None:
+            cmd_repr += e(" env of {!r}").format(self.env)
+        if PY2:
+            return cmd_repr.encode("utf-8")
+        return cmd_repr
+
+    def __unicode__(self):
+        raw = repr(self)
+        if PY2:
+            return raw.decode("utf-8")
+        return raw
+
+
+def clear(app_data):
+    app_data.py_info_clear()
+    _CACHE.clear()
+
+
+___all___ = (
+    "from_exe",
+    "clear",
+    "LogCmd",
+)