Source code for brewtils.decorators

# -*- coding: utf-8 -*-

import functools
import inspect
import json
import os
import sys
import types
from io import open

import requests
import six
import wrapt

    from lark import ParseError
except ImportError:
    from lark.common import ParseError

from brewtils.choices import parse
from brewtils.errors import PluginParamError
from brewtils.models import Command, Parameter, Choices

if sys.version_info.major == 2:
    from funcsigs import signature, Parameter as InspectParameter
    from inspect import signature, Parameter as InspectParameter

__all__ = [

# The wrapt module has a cool feature where you can disable wrapping a decorated function,
# instead just using the original function. This is pretty much exactly what we want - we
# aren't using decorators for their 'real' purpose of wrapping a function, we just want to add
# some metadata to the function object. So we'll disable the wrapping normally, but we need to
# test that enabling the wrapping would work.
_wrap_functions = False

[docs]def system(cls=None, bg_name=None, bg_version=None): """Class decorator that marks a class as a beer-garden System Creates some properties on the class: * ``_commands``: holds all registered commands * ``_name``: an optional system name * ``_version``: an optional system version * ``_current_request``: Reference to the currently executing request Args: cls: The class to decorated bg_name: Optional plugin name bg_version: Optional plugin version Returns: The decorated class """ if cls is None: return functools.partial(system, bg_name=bg_name, bg_version=bg_version) import brewtils.plugin commands = [] for method_name in dir(cls): method = getattr(cls, method_name) method_command = getattr(method, "_command", None) if method_command: commands.append(method_command) cls._commands = commands cls._bg_name = bg_name cls._bg_version = bg_version cls._current_request = property( lambda self: brewtils.plugin.request_context.current_request ) return cls
[docs]def command( _wrapped=None, command_type="ACTION", output_type="STRING", schema=None, form=None, template=None, icon_name=None, description=None, ): """Decorator that marks a function as a beer-garden command For example: .. code-block:: python @command(output_type='JSON') def echo_json(self, message): return message :param _wrapped: The function to decorate. This is handled as a positional argument and shouldn't be explicitly set. :param command_type: The command type. Valid options are Command.COMMAND_TYPES. :param output_type: The output type. Valid options are Command.OUTPUT_TYPES. :param schema: A custom schema definition. :param form: A custom form definition. :param template: A custom template definition. :param icon_name: The icon name. Should be either a FontAwesome or a Glyphicon name. :param description: The command description. Will override the function's docstring. :return: The decorated function. """ if _wrapped is None: return functools.partial( command, command_type=command_type, output_type=output_type, schema=schema, form=form, template=template, icon_name=icon_name, description=description, ) generated_command = _generate_command_from_function(_wrapped) generated_command.command_type = command_type generated_command.output_type = output_type generated_command.icon_name = icon_name if description: generated_command.description = description resolved_mod = _resolve_display_modifiers( _wrapped,, schema=schema, form=form, template=template ) generated_command.schema = resolved_mod["schema"] generated_command.form = resolved_mod["form"] generated_command.template = resolved_mod["template"] func_command = getattr(_wrapped, "_command", None) if func_command: _update_func_command(func_command, generated_command) else: _wrapped._command = generated_command @wrapt.decorator(enabled=_wrap_functions) def wrapper(_double_wrapped, _, _args, _kwargs): return _double_wrapped(*_args, **_kwargs) return wrapper(_wrapped)
[docs]def parameter( _wrapped=None, key=None, type=None, multi=None, display_name=None, optional=None, default=None, description=None, choices=None, nullable=None, maximum=None, minimum=None, regex=None, is_kwarg=None, model=None, form_input_type=None, ): """Decorator that enables Parameter specifications for a beer-garden Command This decorator is intended to be used when more specification is desired for a Parameter. For example:: @parameter(key="message", description="Message to echo", optional=True, type="String", default="Hello, World!") def echo(self, message): return message :param _wrapped: The function to decorate. This is handled as a positional argument and shouldn't be explicitly set. :param key: String specifying the parameter identifier. Must match an argument name of the decorated function. :param type: String indicating the type to use for this parameter. :param multi: Boolean indicating if this parameter is a multi. See documentation for discussion of what this means. :param display_name: String that will be displayed as a label in the user interface. :param optional: Boolean indicating if this parameter must be specified. :param default: The value this parameter will be assigned if not overridden when creating a request. :param description: An additional string that will be displayed in the user interface. :param choices: List or dictionary specifying allowed values. See documentation for more information. :param nullable: Boolean indicating if this parameter is allowed to be null. :param maximum: Integer indicating the maximum value of the parameter. :param minimum: Integer indicating the minimum value of the parameter. :param regex: String describing a regular expression constraint on the parameter. :param is_kwarg: Boolean indicating if this parameter is meant to be part of the decorated function's kwargs. :param model: Class to be used as a model for this parameter. Must be a Python type object, not an instance. :param form_input_type: Only used for string fields. Changes the form input field (e.g. textarea) :return: The decorated function. """ if _wrapped is None: return functools.partial( parameter, key=key, type=type, multi=multi, display_name=display_name, optional=optional, default=default, description=description, choices=choices, nullable=nullable, maximum=maximum, minimum=minimum, regex=regex, is_kwarg=is_kwarg, model=model, form_input_type=form_input_type, ) # Create a command object if one isn't already associated cmd = getattr(_wrapped, "_command", None) if not cmd: cmd = _generate_command_from_function(_wrapped) _wrapped._command = cmd # Every parameter needs a key, so stop that right here if key is None: raise PluginParamError( "Found a parameter definition without a key for " "command '%s'" % ) # If the command doesn't already have a parameter with this key then the # method doesn't have an explicit keyword argument with <key> as the name. # That's only OK if this parameter is meant to be part of the **kwargs. param = cmd.get_parameter_by_key(key) if param is None: if is_kwarg: param = Parameter(key=key, optional=False) cmd.parameters.append(param) else: raise PluginParamError( "Parameter '%s' was not an explicit keyword argument for " "command '%s' and was not marked as part of kwargs (should " "is_kwarg be True?)" % (key, ) # Next, fail if the param is_kwarg=True and the method doesn't have a **kwargs if is_kwarg: kwarg_declared = False for p in signature(_wrapped).parameters.values(): if p.kind == InspectParameter.VAR_KEYWORD: kwarg_declared = True break if not kwarg_declared: raise PluginParamError( "Parameter '%s' of command '%s' was declared as a kwarg argument " "(is_kwarg=True) but the command method does not declare a **kwargs " "argument" % (key, ) # Update parameter definition with the plugin_param arguments param.type = _format_type(param.type if type is None else type) param.multi = param.multi if multi is None else multi param.display_name = param.display_name if display_name is None else display_name param.optional = param.optional if optional is None else optional param.default = param.default if default is None else default param.description = param.description if description is None else description param.choices = param.choices if choices is None else choices param.nullable = param.nullable if nullable is None else nullable param.maximum = param.maximum if maximum is None else maximum param.minimum = param.minimum if minimum is None else minimum param.regex = param.regex if regex is None else regex param.form_input_type = ( param.form_input_type if form_input_type is None else form_input_type ) param.choices = _format_choices(param.choices) # Model is another special case - it requires its own handling if model is not None: param.type = "Dictionary" param.parameters = _generate_nested_params(model) # If the model is not nullable and does not have a default we will try # to generate a one using the defaults defined on the model parameters if not param.nullable and not param.default: param.default = {} for nested_param in param.parameters: if nested_param.default: param.default[nested_param.key] = nested_param.default @wrapt.decorator(enabled=_wrap_functions) def wrapper(_double_wrapped, _, _args, _kwargs): return _double_wrapped(*_args, **_kwargs) return wrapper(_wrapped)
[docs]def parameters(*args): """Specify multiple Parameter definitions at once This can be useful for commands which have a large number of complicated parameters but aren't good candidates for a Model. .. code-block:: python @parameter(**params[cmd1][param1]) @parameter(**params[cmd1][param2]) @parameter(**params[cmd1][param3]) def cmd1(self, **kwargs): pass Can become: .. code-block:: python @parameters(params[cmd1]) def cmd1(self, **kwargs): pass Args: *args (iterable): Positional arguments The first (and only) positional argument must be a list containing dictionaries that describe parameters. Returns: func: The decorated function """ if len(args) == 1: return functools.partial(parameters, args[0]) elif len(args) != 2: raise PluginParamError("@parameters takes a single argument") if not isinstance(args[1], types.FunctionType): raise PluginParamError("@parameters must be applied to a function") try: for param in args[0]: parameter(args[1], **param) except TypeError: raise PluginParamError("@parameters arg must be an iterable of dictionaries") @wrapt.decorator(enabled=_wrap_functions) def wrapper(_double_wrapped, _, _args, _kwargs): return _double_wrapped(*_args, **_kwargs) return wrapper(args[1])
def _update_func_command(func_command, generated_command): """Updates the current function's command with info, (will not override plugin_params)""" = func_command.description = generated_command.description func_command.command_type = generated_command.command_type func_command.output_type = generated_command.output_type func_command.schema = generated_command.schema func_command.form = generated_command.form func_command.template = generated_command.template func_command.icon_name = generated_command.icon_name def _generate_command_from_function(func): """Generates a Command from a function. Uses first line of pydoc as the description.""" # Required for Python 2/3 compatibility if hasattr(func, "func_name"): command_name = func.func_name else: command_name = func.__name__ # Required for Python 2/3 compatibility if hasattr(func, "func_doc"): docstring = func.func_doc else: docstring = func.__doc__ return Command( name=command_name, description=docstring.split("\n")[0] if docstring else None, parameters=_generate_params_from_function(func), ) def _generate_params_from_function(func): """Generate Parameters from function arguments. Will set the Parameter key, default value, and optional value.""" parameters_to_return = [] code = six.get_function_code(func) function_arguments = list(code.co_varnames or [])[: code.co_argcount] function_defaults = list(six.get_function_defaults(func) or []) while len(function_defaults) != len(function_arguments): function_defaults.insert(0, None) for index, param_name in enumerate(function_arguments): # Skip Self or Class reference if index == 0 and isinstance(func, types.FunctionType): continue default = function_defaults[index] optional = False if default is None else True parameters_to_return.append( Parameter(key=param_name, default=default, optional=optional) ) return parameters_to_return def _generate_nested_params(model_class): """Generates Nested Parameters from a Model Class""" parameters_to_return = [] for parameter_definition in model_class.parameters: key = parameter_definition.key parameter_type = parameter_definition.type multi = parameter_definition.multi display_name = parameter_definition.display_name optional = parameter_definition.optional default = parameter_definition.default description = parameter_definition.description nullable = parameter_definition.nullable maximum = parameter_definition.maximum minimum = parameter_definition.minimum regex = parameter_definition.regex choices = _format_choices(parameter_definition.choices) nested_parameters = [] if parameter_definition.parameters: parameter_type = "Dictionary" for nested_class in parameter_definition.parameters: nested_parameters = _generate_nested_params(nested_class) parameters_to_return.append( Parameter( key=key, type=parameter_type, multi=multi, display_name=display_name, optional=optional, default=default, description=description, choices=choices, parameters=nested_parameters, nullable=nullable, maximum=maximum, minimum=minimum, regex=regex, ) ) return parameters_to_return def _resolve_display_modifiers( wrapped, command_name, schema=None, form=None, template=None ): def _load_from_url(url): response = requests.get(url) if response.headers.get("content-type", "").lower() == "application/json": return json.loads(response.text) return response.text def _load_from_path(path): current_dir = os.path.dirname(inspect.getfile(wrapped)) file_path = os.path.abspath(os.path.join(current_dir, path)) with open(file_path, "r") as definition_file: return resolved = {} for key, value in {"schema": schema, "form": form, "template": template}.items(): if isinstance(value, six.string_types): try: if value.startswith("http"): resolved[key] = _load_from_url(value) elif value.startswith("/") or value.startswith("."): loaded_value = _load_from_path(value) resolved[key] = ( loaded_value if key == "template" else json.loads(loaded_value) ) elif key == "template": resolved[key] = value else: raise PluginParamError( "%s specified for command '%s' was not a " "definition, file path, or URL" % (key, command_name) ) except Exception as ex: raise PluginParamError( "Error reading %s definition from '%s' for command " "'%s': %s" % (key, value, command_name, ex) ) elif value is None or (key in ["schema", "form"] and isinstance(value, dict)): resolved[key] = value elif key == "form" and isinstance(value, list): resolved[key] = {"type": "fieldset", "items": value} else: raise PluginParamError( "%s specified for command '%s' was not a definition, " "file path, or URL" % (key, command_name) ) return resolved def _format_type(param_type): if param_type == str: return "String" elif param_type == int: return "Integer" elif param_type == float: return "Float" elif param_type == bool: return "Boolean" elif param_type == dict: return "Dictionary" else: return param_type def _format_choices(choices): def determine_display(display_value): if isinstance(display_value, six.string_types): return "typeahead" return "select" if len(display_value) <= 50 else "typeahead" def determine_type(type_value): if isinstance(type_value, (list, dict)): return "static" elif type_value.startswith("http"): return "url" else: return "command" if not choices: return None if not isinstance(choices, (list, six.string_types, dict)): raise PluginParamError( "Invalid 'choices' provided. Must be a list, dictionary or string." ) elif isinstance(choices, dict): if not choices.get("value"): raise PluginParamError( "No 'value' provided for choices. You must at least " "provide valid values." ) value = choices.get("value") display = choices.get("display", determine_display(value)) choice_type = choices.get("type") strict = choices.get("strict", True) if choice_type is None: choice_type = determine_type(value) elif choice_type not in Choices.TYPES: raise PluginParamError( "Invalid choices type '%s' - Valid type options are %s" % (choice_type, Choices.TYPES) ) else: if ( ( choice_type == "command" and not isinstance(value, (six.string_types, dict)) ) or (choice_type == "url" and not isinstance(value, six.string_types)) or (choice_type == "static" and not isinstance(value, (list, dict))) ): allowed_types = { "command": "('string', 'dictionary')", "url": "('string')", "static": "('list', 'dictionary)", } raise PluginParamError( "Invalid choices value type '%s' - Valid value types for " "choice type '%s' are %s" % (type(value), choice_type, allowed_types[choice_type]) ) if display not in Choices.DISPLAYS: raise PluginParamError( "Invalid choices display '%s' - Valid display options are %s" % (display, Choices.DISPLAYS) ) else: value = choices display = determine_display(value) choice_type = determine_type(value) strict = True # Now parse out type-specific aspects unparsed_value = "" try: if choice_type == "command": if isinstance(value, six.string_types): unparsed_value = value else: unparsed_value = value["command"] details = parse(unparsed_value, parse_as="func") elif choice_type == "url": unparsed_value = value details = parse(unparsed_value, parse_as="url") else: if isinstance(value, dict): unparsed_value = choices.get("key_reference") if unparsed_value is None: raise PluginParamError( "Specifying a static choices dictionary requires a " '"key_reference" field with a reference to another ' 'parameter ("key_reference": "${param_key}")' ) details = {"key_reference": parse(unparsed_value, parse_as="reference")} else: details = {} except ParseError: raise PluginParamError( "Invalid choices definition - Unable to parse '%s'" % unparsed_value ) return Choices( type=choice_type, display=display, value=value, strict=strict, details=details ) # Alias the old names for compatibility command_registrar = system plugin_param = parameter register = command