1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
|
"""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)
|
