| """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) |