1from __future__ import absolute_import
2
3import types
4import six
5from ._util import cached_property
6from ._compat import inspect
7
8__all__ = ('Callable', )
9
10
11_inspect_iscoroutinefunction = getattr(
12 inspect, 'iscoroutinefunction', lambda f: False)
13
14
15class _Reagent(object):
16 pass
17
18
19_reagent = _Reagent()
20
21
22def _f(owner):
23 return owner
24
25
26def _name_binder(descriptor, obj, type):
27 return type, None
28
29
30def _type_binder(descriptor, obj, type):
31 return type, type
32
33
34def _obj_binder(descriptor, obj, type):
35 return obj, obj
36
37
38_descriptor_binder_cache = {}
39
40
41class Descriptor(object):
42
43 def __init__(self, descriptor):
44 self.descriptor = descriptor
45 self.descriptor_class = type(descriptor)
46
47 def detect_function_attr_name(self):
48 indicator = object()
49 descriptor = self.descriptor_class(indicator)
50 for name in dir(descriptor):
51 attr = getattr(descriptor, name)
52 if attr is indicator:
53 # detected!
54 return name
55 else:
56 raise RuntimeError(
57 "The given function doesn't hold the given function as an "
58 "attribute. Is it a correct descriptor?")
59
60 def detect_property(self, obj, type_):
61 d = self.descriptor_class(_f)
62 method_or_value = d.__get__(obj, type_)
63 return method_or_value is obj or method_or_value is type_
64
65 def detect_binder(self, obj, type_):
66 key = (self.descriptor_class, obj is not None)
67 if key not in _descriptor_binder_cache:
68 d = self.descriptor_class(_f)
69 method = d.__get__(obj, type_)
70 if isinstance(method, types.FunctionType):
71 # not a boundmethod - probably staticmethod
72 binder = _name_binder
73 elif method is obj:
74 binder = _obj_binder
75 elif method is type_:
76 binder = _type_binder
77 elif callable(method):
78 owner = method()
79 if owner is type_:
80 binder = _type_binder
81 elif owner is obj:
82 binder = _obj_binder
83 else:
84 binder = None
85 elif method is d:
86 # some non-method descriptor
87 binder = _name_binder
88 else:
89 binder = None
90 if binder is None:
91 raise TypeError(
92 "'descriptor_bind' fails to auto-detect binding rule "
93 "of the given descriptor. Specify the rule by "
94 "'wirerope.wire.descriptor_bind.register'.")
95 _descriptor_binder_cache[key] = binder
96 else:
97 binder = _descriptor_binder_cache[key]
98 return binder
99
100
101class Callable(object):
102 """A wrapper object including more information of callables."""
103
104 def __init__(self, f):
105 self.wrapped_object = f
106 self.is_function_type = type(self) is types.FunctionType # noqa
107 if self.is_descriptor:
108 self.descriptor = Descriptor(f)
109 f = getattr(f, self.descriptor.detect_function_attr_name())
110 else:
111 self.descriptor = None
112 self.wrapped_callable = f
113 self.is_wrapped_coroutine = getattr(f, '_is_coroutine', None)
114 self.is_coroutine = self.is_wrapped_coroutine or \
115 _inspect_iscoroutinefunction(f)
116
117 @cached_property
118 def signature(self):
119 return inspect.signature(self.wrapped_callable)
120
121 @cached_property
122 def parameters(self):
123 return list(self.signature.parameters.values())
124
125 @property
126 def first_parameter(self):
127 return self.parameters[0] if self.parameters else None
128
129 @cached_property
130 def is_boundmethod(self):
131 if self.is_function_type or self.is_builtin_property:
132 return False
133 new_bound = self.wrapped_object.__get__(object())
134 try:
135 if six.PY2:
136 return new_bound is self.wrapped_object
137 else:
138 return type(new_bound) is type(self.wrapped_object) # noqa
139 except Exception:
140 return False
141
142 if six.PY2:
143 @property
144 def is_unboundmethod(self):
145 return type(self.wrapped_object) is type(Callable.__init__) # noqa
146
147 @cached_property
148 def is_descriptor(self):
149 if self.is_boundmethod:
150 return False
151 is_descriptor = type(self.wrapped_object).__get__ \
152 is not types.FunctionType.__get__ # noqa
153 if six.PY2:
154 is_descriptor = is_descriptor and \
155 not (self.is_unboundmethod or self.is_boundmethod)
156 return is_descriptor
157
158 @cached_property
159 def is_builtin_property(self):
160 return issubclass(type(self.wrapped_object), property)
161
162 @cached_property
163 def is_property(self):
164 return self.is_builtin_property or \
165 (self.is_descriptor and self.descriptor.detect_property(
166 _reagent, _Reagent))
167
168 if six.PY34:
169 @cached_property
170 def is_barefunction(self):
171 cc = self.wrapped_callable
172 method_name = cc.__qualname__.split('<locals>.')[-1]
173 if method_name == cc.__name__:
174 return True
175 return False
176 else:
177 @cached_property
178 def is_barefunction(self):
179 # im_class does not exist at this point
180 if self.is_descriptor:
181 return False
182 if six.PY2:
183 if self.is_unboundmethod:
184 return False
185 return not (self.is_membermethod or self.is_classmethod)
186
187 @cached_property
188 def is_member(self):
189 """Test given argument is a method or not.
190
191 :rtype: bool
192
193 :note: The test is partially based on the first parameter name.
194 The test result might be wrong.
195 """
196 if six.PY34:
197 if self.is_barefunction:
198 return False
199 if not self.is_descriptor:
200 return True
201 return self.first_parameter is not None \
202 and self.first_parameter.name == 'self'
203
204 @cached_property
205 def is_membermethod(self):
206 """Test given argument is a method or not.
207
208 :rtype: bool
209
210 :note: The test is partially based on the first parameter name.
211 The test result might be wrong.
212 """
213 if self.is_boundmethod:
214 return False
215 if self.is_property:
216 return False
217 return self.is_member
218
219 @cached_property
220 def is_classmethod(self):
221 """Test given argument is a classmethod or not.
222
223 :rtype: bool
224
225 :note: The test is partially based on the first parameter name.
226 The test result might be wrong.
227 """
228 if isinstance(self.wrapped_object, classmethod):
229 return True
230 if six.PY34:
231 if self.is_barefunction:
232 return False
233 if not self.is_descriptor:
234 return False
235
236 return self.first_parameter is not None \
237 and self.first_parameter.name == 'cls'