mirror of
https://github.com/python/cpython.git
synced 2025-11-03 23:21:29 +00:00
99 lines
3.7 KiB
Python
99 lines
3.7 KiB
Python
"""Convenient client for all PyPI APIs.
|
|
|
|
This module provides a ClientWrapper class which will use the "simple"
|
|
or XML-RPC API to request information or files from an index.
|
|
"""
|
|
|
|
from packaging.pypi import simple, xmlrpc
|
|
|
|
_WRAPPER_MAPPINGS = {'get_release': 'simple',
|
|
'get_releases': 'simple',
|
|
'search_projects': 'simple',
|
|
'get_metadata': 'xmlrpc',
|
|
'get_distributions': 'simple'}
|
|
|
|
_WRAPPER_INDEXES = {'xmlrpc': xmlrpc.Client,
|
|
'simple': simple.Crawler}
|
|
|
|
|
|
def switch_index_if_fails(func, wrapper):
|
|
"""Decorator that switch of index (for instance from xmlrpc to simple)
|
|
if the first mirror return an empty list or raises an exception.
|
|
"""
|
|
def decorator(*args, **kwargs):
|
|
retry = True
|
|
exception = None
|
|
methods = [func]
|
|
for f in wrapper._indexes.values():
|
|
if f != func.__self__ and hasattr(f, func.__name__):
|
|
methods.append(getattr(f, func.__name__))
|
|
for method in methods:
|
|
try:
|
|
response = method(*args, **kwargs)
|
|
retry = False
|
|
except Exception as e:
|
|
exception = e
|
|
if not retry:
|
|
break
|
|
if retry and exception:
|
|
raise exception
|
|
else:
|
|
return response
|
|
return decorator
|
|
|
|
|
|
class ClientWrapper:
|
|
"""Wrapper around simple and xmlrpc clients,
|
|
|
|
Choose the best implementation to use depending the needs, using the given
|
|
mappings.
|
|
If one of the indexes returns an error, tries to use others indexes.
|
|
|
|
:param index: tell which index to rely on by default.
|
|
:param index_classes: a dict of name:class to use as indexes.
|
|
:param indexes: a dict of name:index already instantiated
|
|
:param mappings: the mappings to use for this wrapper
|
|
"""
|
|
|
|
def __init__(self, default_index='simple', index_classes=_WRAPPER_INDEXES,
|
|
indexes={}, mappings=_WRAPPER_MAPPINGS):
|
|
self._projects = {}
|
|
self._mappings = mappings
|
|
self._indexes = indexes
|
|
self._default_index = default_index
|
|
|
|
# instantiate the classes and set their _project attribute to the one
|
|
# of the wrapper.
|
|
for name, cls in index_classes.items():
|
|
obj = self._indexes.setdefault(name, cls())
|
|
obj._projects = self._projects
|
|
obj._index = self
|
|
|
|
def __getattr__(self, method_name):
|
|
"""When asking for methods of the wrapper, return the implementation of
|
|
the wrapped classes, depending the mapping.
|
|
|
|
Decorate the methods to switch of implementation if an error occurs
|
|
"""
|
|
real_method = None
|
|
if method_name in _WRAPPER_MAPPINGS:
|
|
obj = self._indexes[_WRAPPER_MAPPINGS[method_name]]
|
|
real_method = getattr(obj, method_name)
|
|
else:
|
|
# the method is not defined in the mappings, so we try first to get
|
|
# it via the default index, and rely on others if needed.
|
|
try:
|
|
real_method = getattr(self._indexes[self._default_index],
|
|
method_name)
|
|
except AttributeError:
|
|
other_indexes = [i for i in self._indexes
|
|
if i != self._default_index]
|
|
for index in other_indexes:
|
|
real_method = getattr(self._indexes[index], method_name,
|
|
None)
|
|
if real_method:
|
|
break
|
|
if real_method:
|
|
return switch_index_if_fails(real_method, self)
|
|
else:
|
|
raise AttributeError("No index have attribute '%s'" % method_name)
|