Refactor type annotation decorators to use click

This commit rewrites how item_tasks decorators is implemented to use
click primatives rather than custom logic.  The primary advantage is to
the new implementation is that the commandline interface comes with the
item_task annotations for free.  Also, much of the decorating and
conversion logic is handled by click classes directly which
significantly reduces the number of lines of code.

Signed-off-by: Jonathan Beezley <[email protected]>

girder_worker_utils/command.py

@@ -0,0 +1,76 @@
+import json
+
+import click
+
+from .parameter import Parameter
+
+
+class Command(click.Command):
+
+    def __init__(self, *args, **kwargs):
+        params = kwargs.pop('params', None) or []
+        input_params = []
+        output_params = []
+        other_params = []
+        for param in params:
+            if isinstance(param, Parameter) and param.is_input():
+                input_params.append(param)
+            elif isinstance(param, Parameter) and param.is_output():
+                output_params.append(param)
+            else:
+                other_params.append(param)
+        other_params += [
+            click.Option(('--item-tasks-json',), is_flag=True, default=False)
+        ]
+        kwargs['params'] = input_params + other_params
+        self._input_params = input_params
+        self._output_params = output_params
+        self._other_params = other_params
+        super(Command, self).__init__(*args, **kwargs)
+
+    def invoke(self, ctx):
+        if ctx.params.pop('item_tasks_json', False):
+            return self._print_item_tasks_json(ctx)
+        return super(Command, self).invoke(ctx)
+
+    def invoke_item_task(self, bindings):
+        # parse the input bindings to construct an argument list
+        kwargs = {}
+
+        # We need to loop through all of the input_params to get default values
+        # because we are bypassing the part of click that does this by calling
+        # ctx.invoke directly.  This is done because there is no way to
+        # construct CLI argument list from bindings (the name munging logic in
+        # click is strictly one directional).
+        #
+        # Alternatively, we might store the commandline argument name in addition
+        # to the bound variable name so we *can* generate a valid CLI argument
+        # list... this could potentially remove some of the custom logic present
+        # in girder's item_tasks plugin.
+        with self.make_context(self.name, [], resilient_parsing=True) as ctx:
+            for param in self._input_params:
+                kwargs.update(param.get_kwargs_from_input_bindings(bindings, ctx))
+
+            return ctx.invoke(self.callback, **kwargs)
+
+    def item_tasks_json(self, ctx=None):
+        spec = {
+            'name': self.name,
+            'description': self.help or '',
+            'inputs': [],
+            'outputs': []
+        }
+        for param in self._input_params:
+            spec['inputs'].append(param.item_tasks_json(ctx))
+        for param in self._output_params:
+            spec['outputs'].append(param.item_tasks_json(ctx))
+        return spec
+
+    def make_context(self, info_name, args, **kwargs):
+        if '--item-tasks-json' in args:
+            kwargs.setdefault('resilient_parsing', True)
+        return super(Command, self).make_context(info_name, args, **kwargs)
+
+    def _print_item_tasks_json(self, ctx=None):
+        spec = self.item_tasks_json(ctx)
+        click.echo(json.dumps(spec, indent=2))

girder_worker_utils/decorators.py

@@ -1,121 +1,33 @@
-from inspect import getdoc
-try:
-    from inspect import signature
-except ImportError:  # pragma: nocover
-    from funcsigs import signature
-import six
+import click
 
+from .command import Command
+from .input import Input
+from .output import Output
+from .types import String
 
-class MissingDescriptionException(Exception):
-    """Raised when a function is missing description decorators."""
-
-
-class MissingInputException(Exception):
-    """Raised when a required input is missing."""
-
-
-def get_description_attribute(func):
-    """Get the private description attribute from a function."""
-    func = getattr(func, 'run', func)
-    description = getattr(func, '_girder_description', None)
-    if description is None:
-        raise MissingDescriptionException('Function is missing description decorators')
-    return description
-
-
-def argument(name, data_type, *args, **kwargs):
-    """Describe an argument to a function as a function decorator.
-
-    Additional arguments are passed to the type class constructor.
-
-    :param str name: The parameter name from the function declaration
-    :param type: A type class derived from ``girder_worker_utils.types.Base``
-    """
-    if not isinstance(name, six.string_types):
-        raise TypeError('Expected argument name to be a string')
-
-    if callable(data_type):
-        data_type = data_type(name, *args, **kwargs)
-
-    def argument_wrapper(func):
-        func._girder_description = getattr(func, '_girder_description', {})
-        args = func._girder_description.setdefault('arguments', [])
-        sig = signature(func)
-
-        if name not in sig.parameters:
-            raise ValueError('Invalid argument name "%s"' % name)
-
-        data_type.set_parameter(sig.parameters[name], signature=sig)
-        args.insert(0, data_type)
-
-        def call_item_task(inputs, outputs={}):
-            args, kwargs = parse_inputs(func, inputs)
-            return func(*args, **kwargs)
-
-        def describe():
-            return describe_function(func)
-
-        func.call_item_task = call_item_task
-        func.describe = describe
 
+def task(name=None, cls=Command, **attrs):
+    def wrapper(func):
+        cmd_wrapper = click.command(name, cls, **attrs)
+        func.main = cmd_wrapper(func)
+        func.call_item_task = func.main.invoke_item_task
+        func.describe = func.main.item_tasks_json
         return func
+    return wrapper
 
-    return argument_wrapper
-
-
-def describe_function(func):
-    """Return a json description from a decorated function."""
-    description = get_description_attribute(func)
-
-    inputs = [arg.describe() for arg in description.get('arguments', [])]
-    spec = {
-        'name': description.get('name', func.__name__),
-        'inputs': inputs,
-        'mode': 'girder_worker'
-    }
-    desc = description.get('description', getdoc(func))
-    if desc:
-        spec['description'] = desc
-
-    return spec
-
-
-def get_input_data(arg, input_binding):
-    """Parse an input binding from a function argument description.
-
-    :param arg: An instantiated type description
-    :param input_binding: An input binding object
-    :returns: The parameter value
-    """
-    mode = input_binding.get('mode', 'inline')
-    if mode == 'inline' and 'data' in input_binding:
-        value = arg.deserialize(input_binding['data'])
-    elif mode == 'girder':
-        value = input_binding.get('id')
-    else:
-        raise ValueError('Unhandled input mode')
-
-    arg.validate(value)
-    return value
 
+def task_input(*param_decls, **attrs):
+    attrs.setdefault('cls', Input)
+    attrs.setdefault('type', String())
+    return click.option(*param_decls, **attrs)
 
-def parse_inputs(func, inputs):
-    """Parse an object of input bindings from item_tasks.
 
-    :param func: The task function
-    :param dict inputs: The input task bindings object
-    :returns: args and kwargs objects to call the function with
-    """
-    description = get_description_attribute(func)
-    arguments = description.get('arguments', [])
-    args = []
-    kwargs = {}
-    for arg in arguments:
-        desc = arg.describe()
-        input_id = desc['id']
-        name = desc['name']
-        if input_id not in inputs and not arg.has_default():
-            raise MissingInputException('Required input "%s" not provided' % name)
-        if input_id in inputs:
-            kwargs[name] = get_input_data(arg, inputs[input_id])
-    return args, kwargs
+def task_output(*param_decls, **attrs):
+    decls = []
+    for decl in param_decls:
+        if not decl.startswith('-'):
+            decl = '--' + decl
+        decls.append(decl)
+    attrs.setdefault('cls', Output)
+    attrs.setdefault('type', String())
+    return click.argument(*decls, **attrs)

girder_worker_utils/input.py

@@ -0,0 +1,67 @@
+try:
+    from inspect import signature
+except ImportError:
+    from funcsigs import signature
+
+from click.exceptions import BadParameter
+
+from .parameter import Parameter
+
+
+class Input(Parameter):
+    _item_tasks_type = 'input'
+
+    # TODO: Currently, the context argument exists entirely to provide the function
+    #       signature for providing default values.  Click on it's own expects that
+    #       defaults are provided by the @option decorators.  It could be that parsing
+    #       the function signature for default values is a bad idea, and we should
+    #       remove the context argument here.
+    #
+    #       As an additional implementation note, there is some confusion as to
+    #       exactly methods need to have access to the context.  In most cases,
+    #       click uses the context optionally inside it's API.  In the item_tasks
+    #       use case, it is complicated because we cannot create a context (which
+    #       requires the argument list) before parsing the input bindings.  This
+    #       will almost certainly cause pain down the road as code is created that
+    #       *needs* access to the context for special processing.
+    def item_tasks_json(self, ctx=None):
+        spec = super(Input, self).item_tasks_json(ctx)
+
+        # get default from the decorator
+        default = self.get_default(ctx)
+
+        # get default from the function definition
+        if ctx and self.default is None:
+            default = self._get_default_from_function(ctx.command)
+
+        if default is not None:
+            spec['default'] = {
+                'data': default
+            }
+        spec.update(self.type.item_tasks_json(self, ctx))
+        return spec
+
+    def get_kwargs_from_input_bindings(self, bindings, ctx):
+        value = self.default
+        if self.name in bindings:
+            binding = bindings[self.name]
+            if not isinstance(binding, dict) or 'data' not in binding:
+                raise BadParameter('Invalid input binding', ctx=ctx, param=self)
+            value = binding['data']
+
+        if value is None:
+            return {}
+
+        return {self.name: value}
+
+    def _get_default_from_function(self, command):
+        if command is None:
+            return
+        func = command.callback
+        sig = signature(func)
+        if self.name not in sig.parameters:
+            return
+        param = sig.parameters[self.name]
+        if param.default == param.empty:
+            return
+        return param.default

girder_worker_utils/output.py

@@ -0,0 +1,5 @@
+from .parameter import Parameter
+
+
+class Output(Parameter):
+    _item_tasks_type = 'output'

girder_worker_utils/parameter.py

@@ -0,0 +1,20 @@
+import click
+
+
+class Parameter(click.Option):
+    _item_tasks_type = None
+
+    def item_tasks_json(self, command=None):
+        spec = {
+            'id': self.name,
+            'name': self.name,
+            'description': self.help or ''
+        }
+        spec.update(self.type.item_tasks_json(self))
+        return spec
+
+    def is_input(self):
+        return self._item_tasks_type == 'input'
+
+    def is_output(self):
+        return self._item_tasks_type == 'output'