diff options
author | Éric Araujo <merwok@netwok.org> | 2012-06-24 04:07:41 (GMT) |
---|---|---|
committer | Éric Araujo <merwok@netwok.org> | 2012-06-24 04:07:41 (GMT) |
commit | 859aad6a36262383b98ddd45fe3253a882b87ce8 (patch) | |
tree | 1cc50af4fc88c650fe997a2e72f5f26d92a1986c /Lib/packaging/pypi/simple.py | |
parent | dc44f55cc9dc1d016799362c344958baab328ff4 (diff) | |
download | cpython-859aad6a36262383b98ddd45fe3253a882b87ce8.zip cpython-859aad6a36262383b98ddd45fe3253a882b87ce8.tar.gz cpython-859aad6a36262383b98ddd45fe3253a882b87ce8.tar.bz2 |
Remove packaging from the standard library.
Distutils2 will live on on PyPI and be included in the stdlib when it
is ready. See discussion starting at
http://mail.python.org/pipermail/python-dev/2012-June/120430.html
Diffstat (limited to 'Lib/packaging/pypi/simple.py')
-rw-r--r-- | Lib/packaging/pypi/simple.py | 462 |
1 files changed, 0 insertions, 462 deletions
diff --git a/Lib/packaging/pypi/simple.py b/Lib/packaging/pypi/simple.py deleted file mode 100644 index e26d55d..0000000 --- a/Lib/packaging/pypi/simple.py +++ /dev/null @@ -1,462 +0,0 @@ -"""Spider using the screen-scraping "simple" PyPI API. - -This module contains the class Crawler, a simple spider that -can be used to find and retrieve distributions from a project index -(like the Python Package Index), using its so-called simple API (see -reference implementation available at http://pypi.python.org/simple/). -""" - -import http.client -import re -import socket -import sys -import urllib.request -import urllib.parse -import urllib.error -import os - -from fnmatch import translate -from functools import wraps -from packaging import logger -from packaging.metadata import Metadata -from packaging.version import get_version_predicate -from packaging import __version__ as packaging_version -from packaging.pypi.base import BaseClient -from packaging.pypi.dist import (ReleasesList, EXTENSIONS, - get_infos_from_url, MD5_HASH) -from packaging.pypi.errors import (PackagingPyPIError, DownloadError, - UnableToDownload, CantParseArchiveName, - ReleaseNotFound, ProjectNotFound) -from packaging.pypi.mirrors import get_mirrors - -__all__ = ['Crawler', 'DEFAULT_SIMPLE_INDEX_URL'] - -# -- Constants ----------------------------------------------- -DEFAULT_SIMPLE_INDEX_URL = "http://a.pypi.python.org/simple/" -DEFAULT_HOSTS = ("*",) -SOCKET_TIMEOUT = 15 -USER_AGENT = "Python-urllib/%s.%s packaging/%s" % ( - sys.version_info[0], sys.version_info[1], packaging_version) - -# -- Regexps ------------------------------------------------- -EGG_FRAGMENT = re.compile(r'^egg=([-A-Za-z0-9_.]+)$') -HREF = re.compile("""href\\s*=\\s*['"]?([^'"> ]+)""", re.I) -URL_SCHEME = re.compile('([-+.a-z0-9]{2,}):', re.I).match - -# This pattern matches a character entity reference (a decimal numeric -# references, a hexadecimal numeric reference, or a named reference). -ENTITY_SUB = re.compile(r'&(#(\d+|x[\da-fA-F]+)|[\w.:-]+);?').sub -REL = re.compile("""<([^>]*\srel\s*=\s*['"]?([^'">]+)[^>]*)>""", re.I) - - -def socket_timeout(timeout=SOCKET_TIMEOUT): - """Decorator to add a socket timeout when requesting pages on PyPI. - """ - def wrapper(func): - @wraps(func) - def wrapped(self, *args, **kwargs): - old_timeout = socket.getdefaulttimeout() - if hasattr(self, "_timeout"): - timeout = self._timeout - socket.setdefaulttimeout(timeout) - try: - return func(self, *args, **kwargs) - finally: - socket.setdefaulttimeout(old_timeout) - return wrapped - return wrapper - - -def with_mirror_support(): - """Decorator that makes the mirroring support easier""" - def wrapper(func): - @wraps(func) - def wrapped(self, *args, **kwargs): - try: - return func(self, *args, **kwargs) - except DownloadError: - # if an error occurs, try with the next index_url - if self._mirrors_tries >= self._mirrors_max_tries: - try: - self._switch_to_next_mirror() - except KeyError: - raise UnableToDownload("Tried all mirrors") - else: - self._mirrors_tries += 1 - self._projects.clear() - return wrapped(self, *args, **kwargs) - return wrapped - return wrapper - - -class Crawler(BaseClient): - """Provides useful tools to request the Python Package Index simple API. - - You can specify both mirrors and mirrors_url, but mirrors_url will only be - used if mirrors is set to None. - - :param index_url: the url of the simple index to search on. - :param prefer_final: if the version is not mentioned, and the last - version is not a "final" one (alpha, beta, etc.), - pick up the last final version. - :param prefer_source: if the distribution type is not mentioned, pick up - the source one if available. - :param follow_externals: tell if following external links is needed or - not. Default is False. - :param hosts: a list of hosts allowed to be processed while using - follow_externals=True. Default behavior is to follow all - hosts. - :param follow_externals: tell if following external links is needed or - not. Default is False. - :param mirrors_url: the url to look on for DNS records giving mirror - addresses. - :param mirrors: a list of mirrors (see PEP 381). - :param timeout: time in seconds to consider a url has timeouted. - :param mirrors_max_tries": number of times to try requesting informations - on mirrors before switching. - """ - - def __init__(self, index_url=DEFAULT_SIMPLE_INDEX_URL, prefer_final=False, - prefer_source=True, hosts=DEFAULT_HOSTS, - follow_externals=False, mirrors_url=None, mirrors=None, - timeout=SOCKET_TIMEOUT, mirrors_max_tries=0): - super(Crawler, self).__init__(prefer_final, prefer_source) - self.follow_externals = follow_externals - - # mirroring attributes. - parsed = urllib.parse.urlparse(index_url) - self.scheme = parsed[0] - if self.scheme == 'file': - ender = os.path.sep - else: - ender = '/' - if not index_url.endswith(ender): - index_url += ender - # if no mirrors are defined, use the method described in PEP 381. - if mirrors is None: - mirrors = get_mirrors(mirrors_url) - self._mirrors = set(mirrors) - self._mirrors_used = set() - self.index_url = index_url - self._mirrors_max_tries = mirrors_max_tries - self._mirrors_tries = 0 - self._timeout = timeout - - # create a regexp to match all given hosts - self._allowed_hosts = re.compile('|'.join(map(translate, hosts))).match - - # we keep an index of pages we have processed, in order to avoid - # scanning them multple time (eg. if there is multiple pages pointing - # on one) - self._processed_urls = [] - self._projects = {} - - @with_mirror_support() - def search_projects(self, name=None, **kwargs): - """Search the index for projects containing the given name. - - Return a list of names. - """ - if '*' in name: - name.replace('*', '.*') - else: - name = "%s%s%s" % ('*.?', name, '*.?') - name = name.replace('*', '[^<]*') # avoid matching end tag - pattern = ('<a[^>]*>(%s)</a>' % name).encode('utf-8') - projectname = re.compile(pattern, re.I) - matching_projects = [] - - with self._open_url(self.index_url) as index: - index_content = index.read() - - for match in projectname.finditer(index_content): - project_name = match.group(1).decode('utf-8') - matching_projects.append(self._get_project(project_name)) - return matching_projects - - def get_releases(self, requirements, prefer_final=None, - force_update=False): - """Search for releases and return a ReleasesList object containing - the results. - """ - predicate = get_version_predicate(requirements) - if predicate.name.lower() in self._projects and not force_update: - return self._projects.get(predicate.name.lower()) - prefer_final = self._get_prefer_final(prefer_final) - logger.debug('Reading info on PyPI about %s', predicate.name) - self._process_index_page(predicate.name) - - if predicate.name.lower() not in self._projects: - raise ProjectNotFound - - releases = self._projects.get(predicate.name.lower()) - releases.sort_releases(prefer_final=prefer_final) - return releases - - def get_release(self, requirements, prefer_final=None): - """Return only one release that fulfill the given requirements""" - predicate = get_version_predicate(requirements) - release = self.get_releases(predicate, prefer_final)\ - .get_last(predicate) - if not release: - raise ReleaseNotFound("No release matches the given criterias") - return release - - def get_distributions(self, project_name, version): - """Return the distributions found on the index for the specific given - release""" - # as the default behavior of get_release is to return a release - # containing the distributions, just alias it. - return self.get_release("%s (%s)" % (project_name, version)) - - def get_metadata(self, project_name, version): - """Return the metadatas from the simple index. - - Currently, download one archive, extract it and use the PKG-INFO file. - """ - release = self.get_distributions(project_name, version) - if not release.metadata: - location = release.get_distribution().unpack() - pkg_info = os.path.join(location, 'PKG-INFO') - release.metadata = Metadata(pkg_info) - return release - - def _switch_to_next_mirror(self): - """Switch to the next mirror (eg. point self.index_url to the next - mirror url. - - Raise a KeyError if all mirrors have been tried. - """ - self._mirrors_used.add(self.index_url) - index_url = self._mirrors.pop() - # XXX use urllib.parse for a real check of missing scheme part - if not index_url.startswith(("http://", "https://", "file://")): - index_url = "http://%s" % index_url - - if not index_url.endswith("/simple"): - index_url = "%s/simple/" % index_url - - self.index_url = index_url - - def _is_browsable(self, url): - """Tell if the given URL can be browsed or not. - - It uses the follow_externals and the hosts list to tell if the given - url is browsable or not. - """ - # if _index_url is contained in the given URL, we are browsing the - # index, and it's always "browsable". - # local files are always considered browable resources - if self.index_url in url or urllib.parse.urlparse(url)[0] == "file": - return True - elif self.follow_externals: - if self._allowed_hosts(urllib.parse.urlparse(url)[1]): # 1 is netloc - return True - else: - return False - return False - - def _is_distribution(self, link): - """Tell if the given URL matches to a distribution name or not. - """ - #XXX find a better way to check that links are distributions - # Using a regexp ? - for ext in EXTENSIONS: - if ext in link: - return True - return False - - def _register_release(self, release=None, release_info={}): - """Register a new release. - - Both a release or a dict of release_info can be provided, the preferred - way (eg. the quicker) is the dict one. - - Return the list of existing releases for the given project. - """ - # Check if the project already has a list of releases (refering to - # the project name). If not, create a new release list. - # Then, add the release to the list. - if release: - name = release.name - else: - name = release_info['name'] - if name.lower() not in self._projects: - self._projects[name.lower()] = ReleasesList(name, index=self._index) - - if release: - self._projects[name.lower()].add_release(release=release) - else: - name = release_info.pop('name') - version = release_info.pop('version') - dist_type = release_info.pop('dist_type') - self._projects[name.lower()].add_release(version, dist_type, - **release_info) - return self._projects[name.lower()] - - def _process_url(self, url, project_name=None, follow_links=True): - """Process an url and search for distributions packages. - - For each URL found, if it's a download, creates a PyPIdistribution - object. If it's a homepage and we can follow links, process it too. - - :param url: the url to process - :param project_name: the project name we are searching for. - :param follow_links: Do not want to follow links more than from one - level. This parameter tells if we want to follow - the links we find (eg. run recursively this - method on it) - """ - with self._open_url(url) as f: - base_url = f.url - if url not in self._processed_urls: - self._processed_urls.append(url) - link_matcher = self._get_link_matcher(url) - for link, is_download in link_matcher(f.read().decode(), base_url): - if link not in self._processed_urls: - if self._is_distribution(link) or is_download: - self._processed_urls.append(link) - # it's a distribution, so create a dist object - try: - infos = get_infos_from_url(link, project_name, - is_external=self.index_url not in url) - except CantParseArchiveName as e: - logger.warning( - "version has not been parsed: %s", e) - else: - self._register_release(release_info=infos) - else: - if self._is_browsable(link) and follow_links: - self._process_url(link, project_name, - follow_links=False) - - def _get_link_matcher(self, url): - """Returns the right link matcher function of the given url - """ - if self.index_url in url: - return self._simple_link_matcher - else: - return self._default_link_matcher - - def _get_full_url(self, url, base_url): - return urllib.parse.urljoin(base_url, self._htmldecode(url)) - - def _simple_link_matcher(self, content, base_url): - """Yield all links with a rel="download" or rel="homepage". - - This matches the simple index requirements for matching links. - If follow_externals is set to False, dont yeld the external - urls. - - :param content: the content of the page we want to parse - :param base_url: the url of this page. - """ - for match in HREF.finditer(content): - url = self._get_full_url(match.group(1), base_url) - if MD5_HASH.match(url): - yield (url, True) - - for match in REL.finditer(content): - # search for rel links. - tag, rel = match.groups() - rels = [s.strip() for s in rel.lower().split(',')] - if 'homepage' in rels or 'download' in rels: - for match in HREF.finditer(tag): - url = self._get_full_url(match.group(1), base_url) - if 'download' in rels or self._is_browsable(url): - # yield a list of (url, is_download) - yield (url, 'download' in rels) - - def _default_link_matcher(self, content, base_url): - """Yield all links found on the page. - """ - for match in HREF.finditer(content): - url = self._get_full_url(match.group(1), base_url) - if self._is_browsable(url): - yield (url, False) - - @with_mirror_support() - def _process_index_page(self, name): - """Find and process a PyPI page for the given project name. - - :param name: the name of the project to find the page - """ - # Browse and index the content of the given PyPI page. - if self.scheme == 'file': - ender = os.path.sep - else: - ender = '/' - url = self.index_url + name + ender - self._process_url(url, name) - - @socket_timeout() - def _open_url(self, url): - """Open a urllib2 request, handling HTTP authentication, and local - files support. - - """ - scheme, netloc, path, params, query, frag = urllib.parse.urlparse(url) - - # authentication stuff - if scheme in ('http', 'https'): - auth, host = urllib.parse.splituser(netloc) - else: - auth = None - - # add index.html automatically for filesystem paths - if scheme == 'file': - if url.endswith(os.path.sep): - url += "index.html" - - # add authorization headers if auth is provided - if auth: - auth = "Basic " + \ - urllib.parse.unquote(auth).encode('base64').strip() - new_url = urllib.parse.urlunparse(( - scheme, host, path, params, query, frag)) - request = urllib.request.Request(new_url) - request.add_header("Authorization", auth) - else: - request = urllib.request.Request(url) - request.add_header('User-Agent', USER_AGENT) - try: - fp = urllib.request.urlopen(request) - except (ValueError, http.client.InvalidURL) as v: - msg = ' '.join([str(arg) for arg in v.args]) - raise PackagingPyPIError('%s %s' % (url, msg)) - except urllib.error.HTTPError as v: - return v - except urllib.error.URLError as v: - raise DownloadError("Download error for %s: %s" % (url, v.reason)) - except http.client.BadStatusLine as v: - raise DownloadError('%s returned a bad status line. ' - 'The server might be down, %s' % (url, v.line)) - except http.client.HTTPException as v: - raise DownloadError("Download error for %s: %s" % (url, v)) - except socket.timeout: - raise DownloadError("The server timeouted") - - if auth: - # Put authentication info back into request URL if same host, - # so that links found on the page will work - s2, h2, path2, param2, query2, frag2 = \ - urllib.parse.urlparse(fp.url) - if s2 == scheme and h2 == host: - fp.url = urllib.parse.urlunparse( - (s2, netloc, path2, param2, query2, frag2)) - return fp - - def _decode_entity(self, match): - what = match.group(1) - if what.startswith('#x'): - what = int(what[2:], 16) - elif what.startswith('#'): - what = int(what[1:]) - else: - from html.entities import name2codepoint - what = name2codepoint.get(what, match.group(0)) - return chr(what) - - def _htmldecode(self, text): - """Decode HTML entities in the given text.""" - return ENTITY_SUB(self._decode_entity, text) |