Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_call_with_class(
self, method_annotation, request_definition_builder
):
class Class(interfaces.Consumer):
builder = request_definition_builder
method_annotation(Class)
builder = request_definition_builder.method_handler_builder
builder.add_annotation.assert_called_with(
method_annotation, is_class=True
)
def test_static_call_with_class(self, mocker, request_definition_builder):
class Class(interfaces.Consumer):
builder = request_definition_builder
self.FakeMethodAnnotation(Class)
builder = request_definition_builder.method_handler_builder
builder.add_annotation.assert_called_with(mocker.ANY, is_class=True)
def __call__(self, class_or_builder):
if self._is_consumer_class(class_or_builder):
builders = helpers.get_api_definitions(class_or_builder)
builders = filter(self._is_relevant_for_builder, builders)
for name, b in builders:
b.method_handler_builder.add_annotation(self, is_class=True)
helpers.set_api_definition(class_or_builder, name, b)
elif isinstance(class_or_builder, interfaces.RequestDefinitionBuilder):
class_or_builder.method_handler_builder.add_annotation(self)
return class_or_builder
def build(self):
if not self._argument_handler_builder.is_done():
self._auto_fill_remaining_arguments()
argument_handler = self._argument_handler_builder.build()
method_handler = self._method_handler_builder.build()
uri = self._uri.build()
return RequestDefinition(
self._method,
uri,
self._return_type,
argument_handler,
method_handler,
)
class RequestDefinition(interfaces.RequestDefinition):
def __init__(
self, method, uri, return_type, argument_handler, method_handler
):
self._method = method
self._uri = uri
self._return_type = return_type
self._argument_handler = argument_handler
self._method_handler = method_handler
@property
def argument_annotations(self):
return tuple(self._argument_handler.annotations)
@property
def method_annotations(self):
return tuple(self._method_handler.annotations)
"Failed to add `%s` to method `%s`, as all arguments have "
"been annotated."
)
def __init__(self, annotation, func):
self.message = self.message % (annotation, func.__name__)
class ArgumentNotFound(exceptions.AnnotationError):
message = "`%s` does not match any argument name of method `%s`."
def __init__(self, name, func):
self.message = self.message % (name, func.__name__)
class ArgumentAnnotationHandlerBuilder(interfaces.AnnotationHandlerBuilder):
__ANNOTATION_BUILDER_KEY = "#ANNOTATION_BUILDER_KEY#"
@classmethod
def from_func(cls, func):
if not hasattr(func, cls.__ANNOTATION_BUILDER_KEY):
spec = utils.get_arg_spec(func)
handler = cls(func, spec.args)
setattr(func, cls.__ANNOTATION_BUILDER_KEY, handler)
handler.set_annotations(spec.annotations)
return getattr(func, cls.__ANNOTATION_BUILDER_KEY)
def __init__(self, func, arguments, func_is_method=True):
self._arguments = arguments[func_is_method:]
self._annotations = collections.OrderedDict.fromkeys(self._arguments)
self._defined = 0
self._func = func
All attributes are considered, not only defined directly on the class.
Args:
service: A class object.
"""
# In Python 3.3, `inspect.getmembers` doesn't respect the descriptor
# protocol when the first argument is a class. In other words, the
# function includes any descriptors bound to `service` as is rather
# than calling the descriptor's __get__ method. This is seemingly
# fixed in Python 2.7 and 3.4+ (TODO: locate corresponding bug
# report in Python issue tracker). Directly invoking `getattr` to
# force Python's attribute lookup protocol is a decent workaround to
# ensure parity:
class_attributes = ((k, getattr(service, k)) for k in dir(service))
is_definition = interfaces.RequestDefinitionBuilder.__instancecheck__
return [(k, v) for k, v in class_attributes if is_definition(v)]
def __init__(self, missing, path_variables):
missing, path_variables = list(missing), list(path_variables)
self.message = self.message % "', '".join(missing)
if path_variables: # pragma: no cover
self.message += self.implicit_message % "', '".join(path_variables)
class MissingUriVariables(exceptions.InvalidRequestDefinition):
message = "On uri template %s, some variables are not handled: %s"
def __init__(self, uri, remaining_variables):
self.message = self.message % (uri, "', '".join(remaining_variables))
class URIDefinitionBuilder(interfaces.UriDefinitionBuilder):
def __init__(self, uri):
self._uri = uri
self._is_dynamic = False
self._uri_variables = set()
@property
def template(self):
return self._uri
@property
def is_static(self):
return self._uri is not None
@property
def is_dynamic(self):
return self._is_dynamic
def _is_annotation(annotation):
cls = interfaces.Annotation
return utils.is_subclass(annotation, cls) or isinstance(annotation, cls)
# Wrap all definition builders with a special descriptor that
# handles attribute access behavior.
for key, value in namespace.items():
namespace[key] = mcs._wrap_if_definition(name, key, value)
return super(ConsumerMeta, mcs).__new__(mcs, name, bases, namespace)
def __setattr__(cls, key, value):
value = cls._wrap_if_definition(cls.__name__, key, value)
super(ConsumerMeta, cls).__setattr__(key, value)
_Consumer = ConsumerMeta("_Consumer", (), {})
class Consumer(interfaces.Consumer, _Consumer):
"""
Base consumer class with which to define custom consumers.
Example usage:
.. code-block:: python
from uplink import Consumer, get
class GitHub(Consumer):
@get("/users/{user}")
def get_user(self, user):
pass
client = GitHub("https://api.github.com/")
def copy(self):
return self
@property
def _types(self):
types = self._annotations
return ((k, types[k]) for k in types if types[k] is not None)
def build(self):
return ArgumentAnnotationHandler(
self._func, collections.OrderedDict(self._types)
)
class ArgumentAnnotationHandler(interfaces.AnnotationHandler):
def __init__(self, func, arguments):
self._func = func
self._arguments = arguments
@property
def annotations(self):
return iter(self._arguments.values())
def get_relevant_arguments(self, call_args):
annotations = self._arguments
return ((n, annotations[n]) for n in call_args if n in annotations)
def handle_call(self, request_builder, args, kwargs):
call_args = utils.get_call_args(self._func, None, *args, **kwargs)
self.handle_call_args(request_builder, call_args)