mirror of
https://github.com/python/cpython.git
synced 2025-11-04 07:31:38 +00:00
200 lines
8 KiB
Python
200 lines
8 KiB
Python
"""Spider using the XML-RPC PyPI API.
|
|
|
|
This module contains the class Client, a spider that can be used to find
|
|
and retrieve distributions from a project index (like the Python Package
|
|
Index), using its XML-RPC API (see documentation of the reference
|
|
implementation at http://wiki.python.org/moin/PyPiXmlRpc).
|
|
"""
|
|
|
|
import xmlrpc.client
|
|
|
|
from packaging import logger
|
|
from packaging.errors import IrrationalVersionError
|
|
from packaging.version import get_version_predicate
|
|
from packaging.pypi.base import BaseClient
|
|
from packaging.pypi.errors import (ProjectNotFound, InvalidSearchField,
|
|
ReleaseNotFound)
|
|
from packaging.pypi.dist import ReleaseInfo
|
|
|
|
__all__ = ['Client', 'DEFAULT_XMLRPC_INDEX_URL']
|
|
|
|
DEFAULT_XMLRPC_INDEX_URL = 'http://python.org/pypi'
|
|
|
|
_SEARCH_FIELDS = ['name', 'version', 'author', 'author_email', 'maintainer',
|
|
'maintainer_email', 'home_page', 'license', 'summary',
|
|
'description', 'keywords', 'platform', 'download_url']
|
|
|
|
|
|
class Client(BaseClient):
|
|
"""Client to query indexes using XML-RPC method calls.
|
|
|
|
If no server_url is specified, use the default PyPI XML-RPC URL,
|
|
defined in the DEFAULT_XMLRPC_INDEX_URL constant::
|
|
|
|
>>> client = Client()
|
|
>>> client.server_url == DEFAULT_XMLRPC_INDEX_URL
|
|
True
|
|
|
|
>>> client = Client("http://someurl/")
|
|
>>> client.server_url
|
|
'http://someurl/'
|
|
"""
|
|
|
|
def __init__(self, server_url=DEFAULT_XMLRPC_INDEX_URL, prefer_final=False,
|
|
prefer_source=True):
|
|
super(Client, self).__init__(prefer_final, prefer_source)
|
|
self.server_url = server_url
|
|
self._projects = {}
|
|
|
|
def get_release(self, requirements, prefer_final=False):
|
|
"""Return a release with all complete metadata and distribution
|
|
related informations.
|
|
"""
|
|
prefer_final = self._get_prefer_final(prefer_final)
|
|
predicate = get_version_predicate(requirements)
|
|
releases = self.get_releases(predicate.name)
|
|
release = releases.get_last(predicate, prefer_final)
|
|
self.get_metadata(release.name, str(release.version))
|
|
self.get_distributions(release.name, str(release.version))
|
|
return release
|
|
|
|
def get_releases(self, requirements, prefer_final=None, show_hidden=True,
|
|
force_update=False):
|
|
"""Return the list of existing releases for a specific project.
|
|
|
|
Cache the results from one call to another.
|
|
|
|
If show_hidden is True, return the hidden releases too.
|
|
If force_update is True, reprocess the index to update the
|
|
informations (eg. make a new XML-RPC call).
|
|
::
|
|
|
|
>>> client = Client()
|
|
>>> client.get_releases('Foo')
|
|
['1.1', '1.2', '1.3']
|
|
|
|
If no such project exists, raise a ProjectNotFound exception::
|
|
|
|
>>> client.get_project_versions('UnexistingProject')
|
|
ProjectNotFound: UnexistingProject
|
|
|
|
"""
|
|
def get_versions(project_name, show_hidden):
|
|
return self.proxy.package_releases(project_name, show_hidden)
|
|
|
|
predicate = get_version_predicate(requirements)
|
|
prefer_final = self._get_prefer_final(prefer_final)
|
|
project_name = predicate.name
|
|
if not force_update and (project_name.lower() in self._projects):
|
|
project = self._projects[project_name.lower()]
|
|
if not project.contains_hidden and show_hidden:
|
|
# if hidden releases are requested, and have an existing
|
|
# list of releases that does not contains hidden ones
|
|
all_versions = get_versions(project_name, show_hidden)
|
|
existing_versions = project.get_versions()
|
|
hidden_versions = set(all_versions) - set(existing_versions)
|
|
for version in hidden_versions:
|
|
project.add_release(release=ReleaseInfo(project_name,
|
|
version, index=self._index))
|
|
else:
|
|
versions = get_versions(project_name, show_hidden)
|
|
if not versions:
|
|
raise ProjectNotFound(project_name)
|
|
project = self._get_project(project_name)
|
|
project.add_releases([ReleaseInfo(project_name, version,
|
|
index=self._index)
|
|
for version in versions])
|
|
project = project.filter(predicate)
|
|
if len(project) == 0:
|
|
raise ReleaseNotFound("%s" % predicate)
|
|
project.sort_releases(prefer_final)
|
|
return project
|
|
|
|
|
|
def get_distributions(self, project_name, version):
|
|
"""Grab informations about distributions from XML-RPC.
|
|
|
|
Return a ReleaseInfo object, with distribution-related informations
|
|
filled in.
|
|
"""
|
|
url_infos = self.proxy.release_urls(project_name, version)
|
|
project = self._get_project(project_name)
|
|
if version not in project.get_versions():
|
|
project.add_release(release=ReleaseInfo(project_name, version,
|
|
index=self._index))
|
|
release = project.get_release(version)
|
|
for info in url_infos:
|
|
packagetype = info['packagetype']
|
|
dist_infos = {'url': info['url'],
|
|
'hashval': info['md5_digest'],
|
|
'hashname': 'md5',
|
|
'is_external': False,
|
|
'python_version': info['python_version']}
|
|
release.add_distribution(packagetype, **dist_infos)
|
|
return release
|
|
|
|
def get_metadata(self, project_name, version):
|
|
"""Retrieve project metadata.
|
|
|
|
Return a ReleaseInfo object, with metadata informations filled in.
|
|
"""
|
|
# to be case-insensitive, get the informations from the XMLRPC API
|
|
projects = [d['name'] for d in
|
|
self.proxy.search({'name': project_name})
|
|
if d['name'].lower() == project_name]
|
|
if len(projects) > 0:
|
|
project_name = projects[0]
|
|
|
|
metadata = self.proxy.release_data(project_name, version)
|
|
project = self._get_project(project_name)
|
|
if version not in project.get_versions():
|
|
project.add_release(release=ReleaseInfo(project_name, version,
|
|
index=self._index))
|
|
release = project.get_release(version)
|
|
release.set_metadata(metadata)
|
|
return release
|
|
|
|
def search_projects(self, name=None, operator="or", **kwargs):
|
|
"""Find using the keys provided in kwargs.
|
|
|
|
You can set operator to "and" or "or".
|
|
"""
|
|
for key in kwargs:
|
|
if key not in _SEARCH_FIELDS:
|
|
raise InvalidSearchField(key)
|
|
if name:
|
|
kwargs["name"] = name
|
|
projects = self.proxy.search(kwargs, operator)
|
|
for p in projects:
|
|
project = self._get_project(p['name'])
|
|
try:
|
|
project.add_release(release=ReleaseInfo(p['name'],
|
|
p['version'], metadata={'summary': p['summary']},
|
|
index=self._index))
|
|
except IrrationalVersionError as e:
|
|
logger.warning("Irrational version error found: %s", e)
|
|
return [self._projects[p['name'].lower()] for p in projects]
|
|
|
|
def get_all_projects(self):
|
|
"""Return the list of all projects registered in the package index"""
|
|
projects = self.proxy.list_packages()
|
|
for name in projects:
|
|
self.get_releases(name, show_hidden=True)
|
|
|
|
return [self._projects[name.lower()] for name in set(projects)]
|
|
|
|
@property
|
|
def proxy(self):
|
|
"""Property used to return the XMLRPC server proxy.
|
|
|
|
If no server proxy is defined yet, creates a new one::
|
|
|
|
>>> client = Client()
|
|
>>> client.proxy()
|
|
<ServerProxy for python.org/pypi>
|
|
|
|
"""
|
|
if not hasattr(self, '_server_proxy'):
|
|
self._server_proxy = xmlrpc.client.ServerProxy(self.server_url)
|
|
|
|
return self._server_proxy
|