def _make_rule()

in src/python/pants/engine/rules.py [0:0]

• 56 lines of code
• 17 McCabe index (conditional complexity)

def _make_rule(output_type, input_selectors, cacheable=True):
  """A @decorator that declares that a particular static function may be used as a TaskRule.

  As a special case, if the output_type is a subclass of `Goal`, the `Goal.Options` for the `Goal`
  are registered as dependency Optionables.

  :param type output_type: The return/output type for the Rule. This must be a concrete Python type.
  :param list input_selectors: A list of Selector instances that matches the number of arguments
    to the @decorated function.
  """

  is_goal_cls = isinstance(output_type, type) and issubclass(output_type, Goal)
  if is_goal_cls == cacheable:
    raise TypeError('An `@rule` that produces a `Goal` must be declared with @console_rule in order '
                    'to signal that it is not cacheable.')

  def wrapper(func):
    if not inspect.isfunction(func):
      raise ValueError('The @rule decorator must be applied innermost of all decorators.')

    owning_module = sys.modules[func.__module__]
    source = inspect.getsource(func)
    beginning_indent = _get_starting_indent(source)
    if beginning_indent:
      source = "\n".join(line[beginning_indent:] for line in source.split("\n"))
    module_ast = ast.parse(source)

    def resolve_type(name):
      resolved = getattr(owning_module, name, None) or owning_module.__builtins__.get(name, None)
      if resolved is None:
        raise ValueError('Could not resolve type `{}` in top level of module {}'
                         .format(name, owning_module.__name__))
      elif not isinstance(resolved, type):
        raise ValueError('Expected a `type` constructor for `{}`, but got: {} (type `{}`)'
                         .format(name, resolved, type(resolved).__name__))
      return resolved

    gets = OrderedSet()
    rule_func_node = assert_single_element(
      node for node in ast.iter_child_nodes(module_ast)
      if isinstance(node, ast.FunctionDef) and node.name == func.__name__)

    parents_table = {}
    for parent in ast.walk(rule_func_node):
      for child in ast.iter_child_nodes(parent):
        parents_table[child] = parent

    rule_visitor = _RuleVisitor(
      func=func,
      func_node=rule_func_node,
      func_source=source,
      orig_indent=beginning_indent,
      parents_table=parents_table,
    )
    rule_visitor.visit(rule_func_node)
    gets.update(
      Get.create_statically_for_rule_graph(resolve_type(p), resolve_type(s))
      for p, s in rule_visitor.gets)

    # Register dependencies for @console_rule/Goal.
    if is_goal_cls:
      dependency_rules = (optionable_rule(output_type.Options),)
    else:
      dependency_rules = None

    func.rule = TaskRule(
        output_type,
        tuple(input_selectors),
        func,
        input_gets=tuple(gets),
        dependency_rules=dependency_rules,
        cacheable=cacheable,
      )

    return func
  return wrapper