Mercurial > repos > guerler > hhblits

comparison lib/python3.8/site-packages/pip/_internal/self_outdated_check.py @ 0:9e54283cc701 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit d12c32a45bcd441307e632fca6d9af7d60289d44"
author guerler
date Mon, 27 Jul 2020 03:47:31 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:9e54283cc701
1 # The following comment should be removed at some point in the future.
2 # mypy: disallow-untyped-defs=False
3
4 from __future__ import absolute_import
5
6 import datetime
7 import hashlib
8 import json
9 import logging
10 import os.path
11 import sys
12
13 from pip._vendor import pkg_resources
14 from pip._vendor.packaging import version as packaging_version
15 from pip._vendor.six import ensure_binary
16
17 from pip._internal.index.collector import LinkCollector
18 from pip._internal.index.package_finder import PackageFinder
19 from pip._internal.models.search_scope import SearchScope
20 from pip._internal.models.selection_prefs import SelectionPreferences
21 from pip._internal.utils.filesystem import (
22 adjacent_tmp_file,
23 check_path_owner,
24 replace,
25 )
26 from pip._internal.utils.misc import (
27 ensure_dir,
28 get_installed_version,
29 redact_auth_from_url,
30 )
31 from pip._internal.utils.packaging import get_installer
32 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
33
34 if MYPY_CHECK_RUNNING:
35 import optparse
36 from optparse import Values
37 from typing import Any, Dict, Text, Union
38
39 from pip._internal.network.session import PipSession
40
41
42 SELFCHECK_DATE_FMT = "%Y-%m-%dT%H:%M:%SZ"
43
44
45 logger = logging.getLogger(__name__)
46
47
48 def make_link_collector(
49 session, # type: PipSession
50 options, # type: Values
51 suppress_no_index=False, # type: bool
52 ):
53 # type: (...) -> LinkCollector
54 """
55 :param session: The Session to use to make requests.
56 :param suppress_no_index: Whether to ignore the --no-index option
57 when constructing the SearchScope object.
58 """
59 index_urls = [options.index_url] + options.extra_index_urls
60 if options.no_index and not suppress_no_index:
61 logger.debug(
62 'Ignoring indexes: %s',
63 ','.join(redact_auth_from_url(url) for url in index_urls),
64 )
65 index_urls = []
66
67 # Make sure find_links is a list before passing to create().
68 find_links = options.find_links or []
69
70 search_scope = SearchScope.create(
71 find_links=find_links, index_urls=index_urls,
72 )
73
74 link_collector = LinkCollector(session=session, search_scope=search_scope)
75
76 return link_collector
77
78
79 def _get_statefile_name(key):
80 # type: (Union[str, Text]) -> str
81 key_bytes = ensure_binary(key)
82 name = hashlib.sha224(key_bytes).hexdigest()
83 return name
84
85
86 class SelfCheckState(object):
87 def __init__(self, cache_dir):
88 # type: (str) -> None
89 self.state = {} # type: Dict[str, Any]
90 self.statefile_path = None
91
92 # Try to load the existing state
93 if cache_dir:
94 self.statefile_path = os.path.join(
95 cache_dir, "selfcheck", _get_statefile_name(self.key)
96 )
97 try:
98 with open(self.statefile_path) as statefile:
99 self.state = json.load(statefile)
100 except (IOError, ValueError, KeyError):
101 # Explicitly suppressing exceptions, since we don't want to
102 # error out if the cache file is invalid.
103 pass
104
105 @property
106 def key(self):
107 return sys.prefix
108
109 def save(self, pypi_version, current_time):
110 # type: (str, datetime.datetime) -> None
111 # If we do not have a path to cache in, don't bother saving.
112 if not self.statefile_path:
113 return
114
115 # Check to make sure that we own the directory
116 if not check_path_owner(os.path.dirname(self.statefile_path)):
117 return
118
119 # Now that we've ensured the directory is owned by this user, we'll go
120 # ahead and make sure that all our directories are created.
121 ensure_dir(os.path.dirname(self.statefile_path))
122
123 state = {
124 # Include the key so it's easy to tell which pip wrote the
125 # file.
126 "key": self.key,
127 "last_check": current_time.strftime(SELFCHECK_DATE_FMT),
128 "pypi_version": pypi_version,
129 }
130
131 text = json.dumps(state, sort_keys=True, separators=(",", ":"))
132
133 with adjacent_tmp_file(self.statefile_path) as f:
134 f.write(ensure_binary(text))
135
136 try:
137 # Since we have a prefix-specific state file, we can just
138 # overwrite whatever is there, no need to check.
139 replace(f.name, self.statefile_path)
140 except OSError:
141 # Best effort.
142 pass
143
144
145 def was_installed_by_pip(pkg):
146 # type: (str) -> bool
147 """Checks whether pkg was installed by pip
148
149 This is used not to display the upgrade message when pip is in fact
150 installed by system package manager, such as dnf on Fedora.
151 """
152 try:
153 dist = pkg_resources.get_distribution(pkg)
154 return "pip" == get_installer(dist)
155 except pkg_resources.DistributionNotFound:
156 return False
157
158
159 def pip_self_version_check(session, options):
160 # type: (PipSession, optparse.Values) -> None
161 """Check for an update for pip.
162
163 Limit the frequency of checks to once per week. State is stored either in
164 the active virtualenv or in the user's USER_CACHE_DIR keyed off the prefix
165 of the pip script path.
166 """
167 installed_version = get_installed_version("pip")
168 if not installed_version:
169 return
170
171 pip_version = packaging_version.parse(installed_version)
172 pypi_version = None
173
174 try:
175 state = SelfCheckState(cache_dir=options.cache_dir)
176
177 current_time = datetime.datetime.utcnow()
178 # Determine if we need to refresh the state
179 if "last_check" in state.state and "pypi_version" in state.state:
180 last_check = datetime.datetime.strptime(
181 state.state["last_check"],
182 SELFCHECK_DATE_FMT
183 )
184 if (current_time - last_check).total_seconds() < 7 * 24 * 60 * 60:
185 pypi_version = state.state["pypi_version"]
186
187 # Refresh the version if we need to or just see if we need to warn
188 if pypi_version is None:
189 # Lets use PackageFinder to see what the latest pip version is
190 link_collector = make_link_collector(
191 session,
192 options=options,
193 suppress_no_index=True,
194 )
195
196 # Pass allow_yanked=False so we don't suggest upgrading to a
197 # yanked version.
198 selection_prefs = SelectionPreferences(
199 allow_yanked=False,
200 allow_all_prereleases=False, # Explicitly set to False
201 )
202
203 finder = PackageFinder.create(
204 link_collector=link_collector,
205 selection_prefs=selection_prefs,
206 )
207 best_candidate = finder.find_best_candidate("pip").best_candidate
208 if best_candidate is None:
209 return
210 pypi_version = str(best_candidate.version)
211
212 # save that we've performed a check
213 state.save(pypi_version, current_time)
214
215 remote_version = packaging_version.parse(pypi_version)
216
217 local_version_is_older = (
218 pip_version < remote_version and
219 pip_version.base_version != remote_version.base_version and
220 was_installed_by_pip('pip')
221 )
222
223 # Determine if our pypi_version is older
224 if not local_version_is_older:
225 return
226
227 # We cannot tell how the current pip is available in the current
228 # command context, so be pragmatic here and suggest the command
229 # that's always available. This does not accommodate spaces in
230 # `sys.executable`.
231 pip_cmd = "{} -m pip".format(sys.executable)
232 logger.warning(
233 "You are using pip version %s; however, version %s is "
234 "available.\nYou should consider upgrading via the "
235 "'%s install --upgrade pip' command.",
236 pip_version, pypi_version, pip_cmd
237 )
238 except Exception:
239 logger.debug(
240 "There was an error checking the latest version of pip",
241 exc_info=True,
242 )