gwrun argspec

girder_worker_utils/decorators.py

@@ -1,8 +1,9 @@
-from inspect import getdoc
+from inspect import getdoc, cleandoc
 try:
-    from inspect import signature
+    from inspect import signature, Parameter
 except ImportError:  # pragma: nocover
-    from funcsigs import signature
+    from funcsigs import signature, Parameter
+
 import six
 
 
@@ -16,13 +17,190 @@ class MissingInputException(Exception):
 
 def get_description_attribute(func):
     """Get the private description attribute from a function."""
-    func = getattr(func, 'run', func)
-    description = getattr(func, '_girder_description', None)
+    # func = getattr(func, 'run', func)
+    description = getattr(func, GWFuncDesc._func_desc_attr, None)
     if description is None:
         raise MissingDescriptionException('Function is missing description decorators')
     return description
 
 
+class Argument(object):
+    def __init__(self, name, **kwargs):
+        self.name = name
+        for k, v in six.iteritems(kwargs):
+            setattr(self, k, v)
+
+# No default value for this argument
+class Arg(Argument): pass
+# Has a default argument for the value
+class KWArg(Argument):  pass
+class Varargs(Argument): pass
+class Kwargs(Argument): pass
+# class Return(Argument): pass
+
+
+def _clean_function_doc(f):
+    doc = getdoc(f) or ''
+    if isinstance(doc, bytes):
+        doc = doc.decode('utf-8')
+    else:
+        doc = cleandoc(doc)
+    return doc
+
+
+class GWFuncDesc(object):
+    _func_desc_attr = "_gw_function_description"
+    _parameter_repr = ['POSITIONAL_ONLY',
+                       'POSITIONAL_OR_KEYWORD',
+                       'VAR_POSITIONAL',
+                       'KEYWORD_ONLY',
+                       'VAR_KEYWORD']
+
+    @classmethod
+    def get_description(cls, func):
+        # HACK - potentially unwrap celery task
+        # func = getattr(func, 'run', func)
+        if hasattr(func, cls._func_desc_attr) and \
+           isinstance(getattr(func, cls._func_desc_attr), cls):
+            return getattr(func, cls._func_desc_attr)
+        return None
+
+    def __init__(self, func):
+        self.func_name = func.__name__
+        self.func_help = _clean_function_doc(func)
+        self._metadata = {}
+        self._signature = signature(func)
+
+    def __repr__(self):
+        # TODO - make less ugly
+        return "<{}(".format(self.__class__.__name__) + ", ".join(["{}:{}".format(
+            name, self._parameter_repr[self._signature.parameters[name].kind])
+            for name in self._signature.parameters]) + ")>"
+
+    def __getitem__(self, key):
+        return self._construct_argument(
+            self._get_class(self._signature.parameters[key]), key)
+
+    def _construct_argument(self, cls, name, **kwargs):
+        p = self._signature.parameters[name]
+        metadata = {}
+
+        if p.default != p.empty:
+            metadata['default'] = p.default
+        if p.annotation != p.empty:
+            # TODO: make sure annotation is a type and not just garbage
+            metadata['data_type'] = p.annotation
+
+        metadata.update(self._metadata.get(name, {}))
+
+        return cls(name, **metadata)
+
+    def _is_varargs(self, p):
+        return p.kind == Parameter.VAR_POSITIONAL
+
+    def _is_kwargs(self, p):
+        return p.kind == Parameter.VAR_KEYWORD
+
+    def _is_kwarg(self, p):
+        return p.kind == Parameter.KEYWORD_ONLY or (
+            p.kind == Parameter.POSITIONAL_OR_KEYWORD and p.default != p.empty)
+
+    def _is_posarg(self, p):
+        return p.kind == Parameter.POSITIONAL_ONLY or (
+            p.kind == Parameter.POSITIONAL_OR_KEYWORD and p.default == p.empty)
+
+    def _get_class(self, p):
+        if self._is_varargs(p):
+            return Varargs
+        elif self._is_kwargs(p):
+            return Kwargs
+        elif self._is_posarg(p):
+            return Arg
+        elif self._is_kwarg(p):
+            return KWArg
+        else:
+            raise RuntimeError("Could not determine parameter type!")
+
+
+    def set_metadata(self, name, key, value):
+        if name not in self._signature.parameters:
+            raise RuntimeError("{} is not a valid argument to this function!")
+
+        if name not in self._metadata:
+            self._metadata[name] = {}
+
+        self._metadata[name][key] = value
+
+    @property
+    def arguments(self):
+        # Only return arguments if we've declared them as paramaters
+        # This prevents us from returning things like 'self' of bound
+        # methods (e.g. celery tasks) etc.  This is a dubious design
+        # decision.
+        return [
+            self._construct_argument(
+                self._get_class(self._signature.parameters[name]), name)
+            for name in self._signature.parameters if name in self._metadata]
+
+    @property
+    def varargs(self):
+        for name in self._signature.parameters:
+            if name in self._metadata and \
+               self._is_varargs(self._signature.parameters[name]):
+                return self._construct_argument(Varargs, name)
+        return None
+
+    @property
+    def kwargs(self):
+        for name in self._signature.parameters:
+            if name in self._metadata and \
+               self._is_kwargs(self._signature.parameters[name]):
+                return self._construct_argument(Kwargs, name)
+        return None
+
+    @property
+    def positional_args(self):
+        return [arg for arg in self.arguments if isinstance(arg, Arg)]
+
+    @property
+    def keyword_args(self):
+        return [arg for arg in self.arguments if isinstance(arg, KWArg)]
+
+
+def parameter(name, **kwargs):
+    if not isinstance(name, six.string_types):
+        raise TypeError('Expected argument name to be a string')
+
+    data_type = kwargs.get("data_type", None)
+    if data_type is not None and callable(data_type):
+        kwargs['data_type'] = data_type(name, **kwargs)
+
+    def argument_wrapper(func):
+        if not hasattr(func, GWFuncDesc._func_desc_attr):
+            setattr(func, GWFuncDesc._func_desc_attr, GWFuncDesc(func))
+
+        desc = getattr(func, GWFuncDesc._func_desc_attr)
+
+        # Make sure the metadata key exists even if we don't set any
+        # values on it.  This ensures that metadata's keys represent
+        # the full list of parameters that have been identified by the
+        # user (even if there is no actual metadata associated with
+        # the argument).
+        desc._metadata[name] = {}
+
+        for key, value in six.iteritems(kwargs):
+            desc.set_metadata(name, key, value)
+
+        def description():
+            return getattr(func, GWFuncDesc._func_desc_attr)
+
+        func.description = description
+
+        return func
+
+    return argument_wrapper
+
+
 def argument(name, data_type, *args, **kwargs):
     """Describe an argument to a function as a function decorator.
 
@@ -38,8 +216,10 @@ def argument(name, data_type, *args, **kwargs):
         data_type = data_type(name, *args, **kwargs)
 
     def argument_wrapper(func):
-        func._girder_description = getattr(func, '_girder_description', {})
-        args = func._girder_description.setdefault('arguments', [])
+        setattr(func, GWFuncDesc._func_desc_attr,
+                getattr(func, GWFuncDesc._func_desc_attr, {}))
+
+        args = getattr(func, GWFuncDesc._func_desc_attr).setdefault('arguments', [])
         sig = signature(func)
 
         if name not in sig.parameters:

girder_worker_utils/tests/conftest.py

@@ -0,0 +1,5 @@
+import six
+
+collect_ignore = []
+if six.PY2:
+    collect_ignore.append("py3_decorators_test.py")

girder_worker_utils/tests/decorators_test.py

@@ -172,3 +172,141 @@ def test_unhandled_input_binding():
     arg = argument('arg', types.Integer)
     with pytest.raises(ValueError):
         decorators.get_input_data(arg, {})
+
+
+
+###########################
+
+
+import six
+
+from girder_worker_utils.decorators import parameter
+from girder_worker_utils.decorators import (
+    GWFuncDesc,
+    Varargs,
+    Kwargs,
+    Arg,
+    KWArg)
+
+
+def arg(a): pass # noqa
+def varargs(*args): pass # noqa
+def kwarg(a='test'): pass # noqa
+def kwargs(**kwargs): pass # noqa
+def arg_arg(a, b): pass # noqa
+def arg_varargs(a, *args): pass # noqa
+def arg_kwarg(a, b='test'): pass # noqa
+def arg_kwargs(a, **kwargs): pass # noqa
+def kwarg_varargs(a='test', *args): pass # noqa
+def kwarg_kwarg(a='testa', b='testb'): pass # noqa
+def kwarg_kwargs(a='test', **kwargs): pass # noqa
+def arg_kwarg_varargs(a, b='test', *args): pass # noqa
+def arg_kwarg_kwargs(a, b='test', **kwargs): pass # noqa
+def arg_kwarg_varargs_kwargs(a, b='test', *args, **kwargs): pass # noqa
+
+
+
+@pytest.mark.parametrize('func,classes', [
+    (arg, [Arg]),
+    (varargs, [Varargs]),
+    (kwarg, [KWArg]),
+    (kwargs, [Kwargs]),
+    (arg_arg, [Arg, Arg]),
+    (arg_varargs, [Arg, Varargs]),
+    (arg_kwarg, [Arg, KWArg]),
+    (arg_kwargs, [Arg, Kwargs]),
+    (kwarg_varargs, [KWArg, Varargs]),
+    (kwarg_kwarg, [KWArg, KWArg]),
+    (kwarg_kwargs, [KWArg, Kwargs]),
+    (arg_kwarg_varargs, [Arg, KWArg, Varargs]),
+    (arg_kwarg_kwargs, [Arg, KWArg, Kwargs]),
+    (arg_kwarg_varargs_kwargs, [Arg, KWArg, Varargs, Kwargs])
+])
+def test_GWFuncDesc_arguments_returns_expected_classes(func, classes):
+    spec = GWFuncDesc(func)
+    assert len(spec.arguments) == len(classes)
+    for arg, cls in zip(spec.arguments, classes):
+        assert isinstance(arg, cls)
+
+
+no_varargs = [arg, kwarg, kwargs, arg_arg, arg_kwarg,
+              arg_kwargs, kwarg_kwarg, kwarg_kwargs,
+              arg_kwarg_kwargs]
+
+@pytest.mark.parametrize('func', no_varargs)
+def test_GWFuncDesc_varargs_returns_None(func):
+    spec = GWFuncDesc(func)
+    assert spec.varargs is None
+
+
+with_varargs = [varargs, arg_varargs, kwarg_varargs,
+                arg_kwarg_varargs, arg_kwarg_varargs_kwargs]
+
+@pytest.mark.parametrize('func', with_varargs)
+def test_GWFuncDesc_varargs_returns_Vararg(func):
+    spec = GWFuncDesc(func)
+    assert isinstance(spec.varargs, Varargs)
+
+
+@pytest.mark.parametrize('func,names', [
+    (arg, ["a"]),
+    (arg_arg, ["a", "b"]),
+    (arg_varargs, ["a"]),
+    (arg_kwarg, ["a"]),
+    (arg_kwargs, ["a"]),
+    (arg_kwarg_kwargs, ["a"]),
+    (arg_kwarg_varargs_kwargs, ["a"])
+])
+def test_GWFuncDesc_positional_args_correct_names(func, names):
+    spec = GWFuncDesc(func)
+    assert len(spec.positional_args) == len(names)
+    for p, n in zip(spec.positional_args, names):
+        assert isinstance(p, Arg)
+        assert p.name == n
+
+
+# TODO positional_args returns None test
+
+@pytest.mark.parametrize('func,names', [
+    (kwarg, ['a']),
+    (arg_kwarg, ['b']),
+    (kwarg_varargs, ['a']),
+    (kwarg_kwarg, ['a', 'b']),
+    (kwarg_kwargs, ['a']),
+    (arg_kwarg_varargs, ['b']),
+    (arg_kwarg_kwargs, ['b']),
+    (arg_kwarg_varargs_kwargs, ['b']),
+])
+def test_GWFuncDesc_keyword_args_correct_names(func, names):
+    spec = GWFuncDesc(func)
+    assert len(spec.keyword_args) == len(names)
+    for p, n in zip(spec.keyword_args, names):
+        assert isinstance(p, KWArg)
+        assert p.name == n
+
+# TODO keyword_args returns None test
+@pytest.mark.parametrize('func,defaults', [
+    (kwarg, ['test']),
+    (arg_kwarg, ['test']),
+    (kwarg_varargs, ['test']),
+    (kwarg_kwarg, ['testa', 'testb']),
+    (kwarg_kwargs, ['test']),
+    (arg_kwarg_varargs, ['test']),
+    (arg_kwarg_kwargs, ['test']),
+    (arg_kwarg_varargs_kwargs, ['test']),
+])
+def test_GWFuncDesc_keyword_args_have_defaults(func, defaults):
+    spec = GWFuncDesc(func)
+    assert len(spec.keyword_args) == len(defaults)
+    for p, d in zip(spec.keyword_args, defaults):
+        assert hasattr(p, 'default')
+        assert p.default == d
+
+
+def test_parameter_decorator_adds_metadata():
+    @parameter('a', test='TEST')
+    def arg(a):
+        pass
+
+    assert hasattr(arg._girder_spec['a'], 'test')
+    assert arg._girder_spec['a'].test == 'TEST'