Parse selectors

This doesn't use hologram, as it needs more permissive schemas than what hologram provides.
Added some unit tests

core/dbt/contracts/graph/unparsed.py

@@ -1,5 +1,9 @@
 from dbt.node_types import NodeType
-from dbt.contracts.util import Replaceable, Mergeable
+from dbt.contracts.util import (
+    AdditionalPropertiesMixin,
+    Mergeable,
+    Replaceable,
+)
 # trigger the PathEncoder
 import dbt.helper_types  # noqa:F401
 from dbt.exceptions import CompilationException
@@ -177,32 +181,12 @@ def __bool__(self):
 
 
 @dataclass
-class AdditionalPropertiesAllowed(ExtensibleJsonSchemaMixin):
+class AdditionalPropertiesAllowed(
+    AdditionalPropertiesMixin,
+    ExtensibleJsonSchemaMixin
+):
     _extra: Dict[str, Any] = field(default_factory=dict)
 
-    @property
-    def extra(self):
-        return self._extra
-
-    @classmethod
-    def from_dict(cls, data, validate=True):
-        self = super().from_dict(data=data, validate=validate)
-        keys = self.to_dict(validate=False, omit_none=False)
-        for key, value in data.items():
-            if key not in keys:
-                self._extra[key] = value
-        return self
-
-    def to_dict(self, omit_none=True, validate=False):
-        data = super().to_dict(omit_none=omit_none, validate=validate)
-        data.update(self._extra)
-        return data
-
-    def replace(self, **kwargs):
-        dct = self.to_dict(omit_none=False, validate=False)
-        dct.update(kwargs)
-        return self.from_dict(dct)
-
 
 @dataclass
 class ExternalPartition(AdditionalPropertiesAllowed, Replaceable):

core/dbt/contracts/selection.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass, field
+from hologram import JsonSchemaMixin
+
+from typing import List, Dict, Any, Union
+
+
+@dataclass
+class SelectorDefinition(JsonSchemaMixin):
+    name: str
+    definition: Union[str, Dict[str, Any]]
+
+
+@dataclass
+class SelectorFile(JsonSchemaMixin):
+    selectors: List[SelectorDefinition]
+    version: int = 2
+
+
+@dataclass
+class SelectorCollection:
+    packages: Dict[str, List[SelectorFile]] = field(default_factory=dict)

core/dbt/contracts/util.py

@@ -44,3 +44,35 @@ def merged(self, *args):
 class Writable:
     def write(self, path: str, omit_none: bool = False):
         write_json(path, self.to_dict(omit_none=omit_none))  # type: ignore
+
+
+class AdditionalPropertiesMixin:
+    """Make this class an extensible property.
+
+    The underlying class definition must include a type definition for a field
+    named '_extra' that is of type `Dict[str, Any]`.
+    """
+    ADDITIONAL_PROPERTIES = True
+
+    @classmethod
+    def from_dict(cls, data, validate=True):
+        self = super().from_dict(data=data, validate=validate)
+        keys = self.to_dict(validate=False, omit_none=False)
+        for key, value in data.items():
+            if key not in keys:
+                self.extra[key] = value
+        return self
+
+    def to_dict(self, omit_none=True, validate=False):
+        data = super().to_dict(omit_none=omit_none, validate=validate)
+        data.update(self.extra)
+        return data
+
+    def replace(self, **kwargs):
+        dct = self.to_dict(omit_none=False, validate=False)
+        dct.update(kwargs)
+        return self.from_dict(dct)
+
+    @property
+    def extra(self):
+        return self._extra

core/dbt/graph/cli.py

@@ -2,9 +2,12 @@
 import itertools
 
 from typing import (
-    List, Optional
+    Dict, List, Optional, Tuple, Any, Union
 )
 
+from dbt.contracts.selection import SelectorDefinition
+from dbt.exceptions import InternalException, ValidationException
+
 from .selector_spec import (
     SelectionUnion,
     SelectionSpec,
@@ -97,3 +100,153 @@ def parse_test_selectors(
     return SelectionIntersection(
         components=[base, intersect_with], expect_exists=True
     )
+
+
+def _get_list_dicts(
+    dct: Dict[str, Any], key: str
+) -> List[Union[str, Dict[str, Any]]]:
+    result: List[Dict[str, Any]] = []
+    if key not in dct:
+        raise InternalException(
+            f'Expected to find key {key} in dict, only found {list(dct)}'
+        )
+    values = dct[key]
+    if not isinstance(values, list):
+        raise ValidationException(
+            f'Invalid value type {type(values)} in key "{key}" '
+            f'(value "{values}")'
+        )
+    for value in values:
+        if isinstance(value, dict):
+            for value_key in value:
+                if not isinstance(value_key, str):
+                    raise ValidationException(
+                        f'Expected all keys to "{key}" dict to be strings, '
+                        f'but "{value_key}" is a "{type(value_key)}"'
+                    )
+            result.append(value)
+        elif isinstance(value, str):
+            result.append(value)
+        else:
+            raise ValidationException(
+                f'Invalid value type {type(value)} in key "{key}", expected dict '
+                f'or str (value: {value}).'
+            )
+
+    return result
+
+
+def _parse_include_exclude_subdefs(
+    definitions: List[Dict[str, Any]]
+) -> Tuple[List[SelectionSpec], Optional[SelectionSpec]]:
+    include_parts: List[SelectionSpec] = []
+    exclusions: Optional[List[Dict[str, Any]]] = None
+
+    for definition in definitions:
+        if 'exclude' in definition:
+            # do not allow multiple exclude: defs at the same level
+            if exclusions is not None:
+                raise ValidationException(
+                    f'Got multiple exclusion definitions in definition list '
+                    f'{definitions}'
+                )
+            exclusions = _get_list_dicts(definition, 'exclude')
+        else:
+            include_parts.append(parse_from_definition(definition))
+
+    diff_arg: Optional[SelectionSpec] = None
+    if exclusions:
+        parsed_exclusions = [
+            parse_from_definition(excl) for excl in exclusions
+        ]
+        if len(exclusions) == 1:
+            diff_arg = parsed_exclusions[0]
+        else:
+            diff_arg = SelectionUnion(
+                components=parsed_exclusions,
+                raw=exclusions
+            )
+    return (include_parts, diff_arg)
+
+
+def parse_union_definition(definition: Dict[str, Any]) -> SelectionSpec:
+    union_def_parts = _get_list_dicts(definition, 'union')
+    include, exclude = _parse_include_exclude_subdefs(union_def_parts)
+
+    union = SelectionUnion(components=include)
+
+    if exclude is None:
+        union.raw = definition
+        return union
+    else:
+        return SelectionDifference(
+            components=[union, exclude],
+            raw=definition
+        )
+
+
+def parse_intersection_definition(
+    definition: Dict[str, Any]
+) -> SelectionSpec:
+    intersection_def_parts = _get_list_dicts(definition, 'intersection')
+    include, exclude = _parse_include_exclude_subdefs(intersection_def_parts)
+    intersection = SelectionIntersection(components=include)
+    if exclude is None:
+        intersection.raw = definition
+        return intersection
+    else:
+        return SelectionDifference(
+            components=[intersection, exclude],
+            raw=definition
+        )
+
+
+def parse_dict_definition(definition: Dict[str, Any]) -> SelectionSpec:
+    if len(definition) == 1:
+        key = list(definition)[0]
+        value = definition[key]
+        if not isinstance(key, str):
+            raise ValidationException(
+                f'Expected definition key to be a "str", got one of type '
+                f'"{type(key)}" ({key})'
+            )
+        dct = {
+            'method': key,
+            'value': value,
+        }
+    elif 'method' in definition and 'value' in definition:
+        dct = definition
+    else:
+        raise ValidationException(
+            f'Expected exactly 1 key in the selection definition or "method" '
+            f'and "value" keys, but got {list(definition)}'
+        )
+
+    # if key isn't a valid method name, this will raise
+    return SelectionCriteria.from_dict(dct, dct)
+
+
+def parse_from_definition(definition: Dict[str, Any]) -> SelectionSpec:
+    if 'union' in definition:
+        return parse_union_definition(definition)
+    elif 'intersection' in definition:
+        return parse_intersection_definition(definition)
+    elif isinstance(definition, dict):
+        return parse_dict_definition(definition)
+    elif isinstance(definition, str):
+        return SelectionCriteria.from_single_spec(definition)
+    else:
+        raise ValidationException(
+            f'Expected to find str or dict, instead found '
+            f'{type(definition)}: {definition}'
+        )
+
+
+def parse_from_selectors_definition(
+    selectors: List[SelectorDefinition]
+) -> Dict[str, SelectionSpec]:
+    result: Dict[str, SelectionSpec] = {}
+    selector: SelectorDefinition
+    for selector in selectors:
+        result[selector.name] = parse_from_definition(selector.definition)
+    return result

core/dbt/graph/selector.py

@@ -93,15 +93,15 @@ def collect_specified_neighbors(
         overlap with the selected set).
         """
         additional: Set[UniqueId] = set()
-        if spec.select_childrens_parents:
+        if spec.childrens_parents:
             additional.update(self.graph.select_childrens_parents(selected))
 
-        if spec.select_parents:
-            depth = spec.select_parents_max_depth
+        if spec.parents:
+            depth = spec.parents_depth
             additional.update(self.graph.select_parents(selected, depth))
 
-        if spec.select_children:
-            depth = spec.select_children_max_depth
+        if spec.children:
+            depth = spec.children_depth
             additional.update(self.graph.select_children(selected, depth))
         return additional
 

core/dbt/graph/selector_methods.py

@@ -23,6 +23,7 @@
     InternalException,
     RuntimeException,
 )
+from dbt.node_types import NodeType
 
 
 SELECTOR_GLOB = '*'
@@ -38,6 +39,7 @@ class MethodName(StrEnum):
     Config = 'config'
     TestName = 'test_name'
     TestType = 'test_type'
+    ResourceType = 'resource_type'
 
 
 def is_selected_node(real_node, node_selector):
@@ -259,7 +261,9 @@ def __eq__(self, other):
 
 class ConfigSelectorMethod(SelectorMethod):
     def search(
-        self, included_nodes: Set[UniqueId], selector: str
+        self,
+        included_nodes: Set[UniqueId],
+        selector: Any,
     ) -> Iterator[UniqueId]:
         parts = self.arguments
         # special case: if the user wanted to compare test severity,
@@ -276,14 +280,25 @@ def search(
             except AttributeError:
                 continue
             else:
-                # the selector can only be a str, so call str() on the value.
-                # of course, if one wished to render the selector in the jinja
-                # native env, this would no longer be true
-
-                if selector == str(value):
+                if selector == value:
                     yield node
 
 
+class ResourceTypeSelectorMethod(SelectorMethod):
+    def search(
+        self, included_nodes: Set[UniqueId], selector: str
+    ) -> Iterator[UniqueId]:
+        try:
+            resource_type = NodeType(selector)
+        except ValueError as exc:
+            raise RuntimeException(
+                f'Invalid resource_type selector "{selector}"'
+            ) from exc
+        for node, real_node in self.parsed_nodes(included_nodes):
+            if real_node.resource_type == resource_type:
+                yield node
+
+
 class TestNameSelectorMethod(SelectorMethod):
     def search(
         self, included_nodes: Set[UniqueId], selector: str