[component] remove DefsModuleComponent (#29189)

The wrapped components were confusing me so I took a spin at refactoring
to try to simplify things.

`get_component()` for (maybe) getting a `Component` at `context.path` 
`DefsFolderComponent.get()` for expecting to get a `DefsFolderComponent`
like for `defs` folder

## How I Tested These Changes

existing coverage

Author: alangenfeld

python_modules/dagster/dagster/components/core/defs_module.py

@@ -1,6 +1,5 @@
 import inspect
-from abc import abstractmethod
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterator, Mapping, Sequence
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Optional, TypeVar
@@ -45,34 +44,33 @@ class ComponentFileModel(BaseModel):
     requirements: Optional[ComponentRequirementsModel] = None
 
 
-#########
-# MODULES
-#########
-
-
-@dataclass
-class DefsModuleComponent(Component):
-    path: Path
-
-    @classmethod
-    def from_context(cls, context: ComponentLoadContext) -> Optional["DefsModuleComponent"]:
-        """Attempts to load a component from the given context. Iterates through potential component
-        type matches, prioritizing more specific types: YAML, Python, plain Dagster defs, and component
-        folder.
-        """
-        # this defines the priority of the module types
-        module_types = (
-            WrappedYamlComponent,
-            WrappedPythonicComponent,
-            DagsterDefsComponent,
-            DefsFolderComponent,
+def get_component(context: ComponentLoadContext) -> Optional[Component]:
+    """Attempts to load a component from the given context. Iterates through potential component
+    type matches, prioritizing more specific types: YAML, Python, plain Dagster defs, and component
+    folder.
+    """
+    # in priority order
+    # yaml component
+    if (context.path / "component.yaml").exists():
+        return load_yaml_component(context)
+    # pythonic component
+    elif (context.path / "component.py").exists():
+        return load_pythonic_component(context)
+    # defs
+    elif (context.path / "definitions.py").exists():
+        return DagsterDefsComponent(path=context.path / "definitions.py")
+    elif context.path.suffix == ".py":
+        return DagsterDefsComponent(path=context.path)
+    # folder
+    elif context.path.is_dir():
+        children = _crawl(context)
+        return DefsFolderComponent(
+            path=context.path,
+            children=children,
+            asset_post_processors=None,
         )
-        module_filter = filter(None, (cls.from_context(context) for cls in module_types))
-        return next(module_filter, None)
 
-    @classmethod
-    def load(cls, attributes: Any, context: ComponentLoadContext) -> "DefsModuleComponent":
-        return check.not_none(cls.from_context(context))
+    return None
 
 
 @dataclass
@@ -81,15 +79,16 @@ class DefsFolderComponentYamlSchema(Resolvable):
 
 
 @dataclass
-class DefsFolderComponent(DefsModuleComponent):
+class DefsFolderComponent(Component):
     """A folder which may contain multiple submodules, each
     which define components.
 
     Optionally enables postprocessing to modify the Dagster definitions
     produced by submodules.
     """
 
-    submodules: Sequence[DefsModuleComponent]
+    path: Path
+    children: Mapping[Path, Component]
     asset_post_processors: Optional[Sequence[AssetPostProcessor]]
 
     @classmethod
@@ -105,59 +104,54 @@ def load(cls, attributes: Any, context: ComponentLoadContext) -> "DefsFolderComp
             context.resolution_context.at_path("attributes"),
             attributes,
         )
-        submodules = check.not_none(cls.submodules_from_context(context))
+
         return DefsFolderComponent(
             path=context.path,
-            submodules=submodules,
+            children=_crawl(context),
             asset_post_processors=resolved_attributes.asset_post_processors,
         )
 
-    @classmethod
-    def submodules_from_context(
-        cls, context: ComponentLoadContext
-    ) -> Optional[Sequence[DefsModuleComponent]]:
-        if not context.path.is_dir():
-            return None
-        submodules = (
-            DefsModuleComponent.from_context(context.for_path(subpath))
-            for subpath in context.path.iterdir()
-        )
-        return list(filter(None, submodules))
-
-    @classmethod
-    def from_context(cls, context: ComponentLoadContext) -> Optional["DefsFolderComponent"]:
-        submodules = cls.submodules_from_context(context)
-        if submodules is None:
-            return None
-        return DefsFolderComponent(
-            path=context.path, submodules=submodules, asset_post_processors=None
-        )
-
     def build_defs(self, context: ComponentLoadContext) -> Definitions:
         defs = Definitions.merge(
-            *(
-                submodule.build_defs(context.for_path(submodule.path))
-                for submodule in self.submodules
-            )
+            *(child.build_defs(context.for_path(path)) for path, child in self.children.items())
         )
         for post_processor in self.asset_post_processors or []:
             defs = post_processor(defs)
         return defs
 
+    @classmethod
+    def get(cls, context: ComponentLoadContext) -> "DefsFolderComponent":
+        component = get_component(context)
+        return check.inst(
+            component,
+            DefsFolderComponent,
+            f"Expected DefsFolderComponent at {context.path}, got {component}.",
+        )
+
+    def iterate_components(self) -> Iterator[Component]:
+        for component in self.children.values():
+            if isinstance(component, DefsFolderComponent):
+                yield from component.iterate_components()
+
+            yield component
+
+
+def _crawl(context: ComponentLoadContext) -> Mapping[Path, Component]:
+    found = {}
+    for subpath in context.path.iterdir():
+        component = get_component(context.for_path(subpath))
+        if component:
+            found[subpath] = component
+    return found
+
 
 @dataclass
-class DagsterDefsComponent(DefsModuleComponent):
+class DagsterDefsComponent(Component):
     """A Python module containing Dagster definitions. Used for implicit loading of
     Dagster definitions from Python files in the defs folder.
     """
 
-    @classmethod
-    def from_context(cls, context: ComponentLoadContext) -> Optional["DagsterDefsComponent"]:
-        if (context.path / "definitions.py").exists():
-            return DagsterDefsComponent(path=context.path / "definitions.py")
-        elif context.path.suffix == ".py":
-            return DagsterDefsComponent(path=context.path)
-        return None
+    path: Path
 
     def build_defs(self, context: ComponentLoadContext) -> Definitions:
         module = context.load_defs_relative_python_module(self.path)
@@ -173,103 +167,57 @@ def build_defs(self, context: ComponentLoadContext) -> Definitions:
             )
 
 
-@dataclass
-class WrappedDefsModuleComponent(DefsModuleComponent):
-    """A module containing a component definition. ABC implemented by
-    WrappedPythonicComponent and WrappedYamlComponent.
-    """
-
-    wrapped: Component
-
-    @staticmethod
-    @abstractmethod
-    def get_component_def_path(path: Path) -> Path: ...
-
-    @classmethod
-    @abstractmethod
-    def get_component(cls, context: ComponentLoadContext) -> Component: ...
-
-    @classmethod
-    def from_context(cls, context: ComponentLoadContext) -> Optional["WrappedDefsModuleComponent"]:
-        if cls.get_component_def_path(context.path).exists():
-            return cls(path=context.path, wrapped=cls.get_component(context))
-        return None
-
-    def build_defs(self, context: ComponentLoadContext) -> Definitions:
-        return self.wrapped.build_defs(context)
-
-
-@dataclass
-class WrappedPythonicComponent(WrappedDefsModuleComponent):
-    """A module which contains a component definition specified in Python
-    with an @component decorator.
-    """
-
-    @staticmethod
-    def get_component_def_path(path: Path) -> Path:
-        return path / "component.py"
-
-    @classmethod
-    def get_component(cls, context) -> Component:
-        module = context.load_defs_relative_python_module(cls.get_component_def_path(context.path))
-        component_loaders = list(inspect.getmembers(module, is_component_loader))
-        if len(component_loaders) == 0:
-            raise DagsterInvalidDefinitionError("No component loaders found in module")
-        elif len(component_loaders) == 1:
-            _, component_loader = component_loaders[0]
-            return WrappedPythonicComponent(path=context.path, wrapped=component_loader(context))
-        else:
-            raise DagsterInvalidDefinitionError(
-                f"Multiple component loaders found in module: {component_loaders}"
-            )
-
-
-@dataclass
-class WrappedYamlComponent(WrappedDefsModuleComponent):
-    """A module containing a component definition specified in a component.yaml file."""
+def load_pythonic_component(context: ComponentLoadContext) -> Component:
+    module = context.load_defs_relative_python_module(context.path / "component.py")
+    component_loaders = list(inspect.getmembers(module, is_component_loader))
+    if len(component_loaders) == 0:
+        raise DagsterInvalidDefinitionError("No component loaders found in module")
+    elif len(component_loaders) == 1:
+        _, component_loader = component_loaders[0]
+        return component_loader(context)
+    else:
+        raise DagsterInvalidDefinitionError(
+            f"Multiple component loaders found in module: {component_loaders}"
+        )
 
-    @staticmethod
-    def get_component_def_path(path: Path) -> Path:
-        return path / "component.yaml"
 
-    @classmethod
-    def get_component(cls, context: ComponentLoadContext) -> Component:
-        # parse the yaml file
-        component_def_path = cls.get_component_def_path(context.path)
-        source_tree = parse_yaml_with_source_positions(
-            component_def_path.read_text(), str(component_def_path)
-        )
-        component_file_model = _parse_and_populate_model_with_annotated_errors(
-            cls=ComponentFileModel, obj_parse_root=source_tree, obj_key_path_prefix=[]
+def load_yaml_component(context: ComponentLoadContext) -> Component:
+    # parse the yaml file
+    component_def_path = context.path / "component.yaml"
+    source_tree = parse_yaml_with_source_positions(
+        component_def_path.read_text(), str(component_def_path)
+    )
+    component_file_model = _parse_and_populate_model_with_annotated_errors(
+        cls=ComponentFileModel, obj_parse_root=source_tree, obj_key_path_prefix=[]
+    )
+
+    # find the component type
+    type_str = context.normalize_component_type_str(component_file_model.type)
+    key = PluginObjectKey.from_typename(type_str)
+    obj = load_package_object(key)
+    if not isinstance(obj, type) or not issubclass(obj, Component):
+        raise DagsterInvalidDefinitionError(
+            f"Component type {type_str} is of type {type(obj)}, but must be a subclass of dagster.Component"
         )
 
-        # find the component type
-        type_str = context.normalize_component_type_str(component_file_model.type)
-        key = PluginObjectKey.from_typename(type_str)
-        obj = load_package_object(key)
-        if not isinstance(obj, type) or not issubclass(obj, Component):
-            raise DagsterInvalidDefinitionError(
-                f"Component type {type_str} is of type {type(obj)}, but must be a subclass of dagster.Component"
-            )
-
-        model_cls = obj.get_model_cls()
-        context = context.with_rendering_scope(
-            obj.get_additional_scope()
-        ).with_source_position_tree(source_tree.source_position_tree)
-
-        # grab the attributes from the yaml file
-        with pushd(str(context.path)):
-            if model_cls is None:
-                attributes = None
-            elif source_tree:
-                attributes_position_tree = source_tree.source_position_tree.children["attributes"]
-                with enrich_validation_errors_with_source_position(
-                    attributes_position_tree, ["attributes"]
-                ):
-                    attributes = TypeAdapter(model_cls).validate_python(
-                        component_file_model.attributes
-                    )
-            else:
+    model_cls = obj.get_model_cls()
+    context = context.with_rendering_scope(
+        obj.get_additional_scope(),
+    ).with_source_position_tree(
+        source_tree.source_position_tree,
+    )
+
+    # grab the attributes from the yaml file
+    with pushd(str(context.path)):
+        if model_cls is None:
+            attributes = None
+        elif source_tree:
+            attributes_position_tree = source_tree.source_position_tree.children["attributes"]
+            with enrich_validation_errors_with_source_position(
+                attributes_position_tree, ["attributes"]
+            ):
                 attributes = TypeAdapter(model_cls).validate_python(component_file_model.attributes)
+        else:
+            attributes = TypeAdapter(model_cls).validate_python(component_file_model.attributes)
 
-        return obj.load(attributes, context)
+    return obj.load(attributes, context)

python_modules/dagster/dagster/components/core/load_defs.py

@@ -39,13 +39,13 @@ def load_defs(defs_root: ModuleType) -> Definitions:
         resources (Optional[Mapping[str, object]]): A mapping of resource keys to resources
             to apply to the definitions.
     """
-    from dagster.components.core.defs_module import DefsModuleComponent
+    from dagster.components.core.defs_module import get_component
     from dagster.components.core.package_entry import discover_entry_point_package_objects
     from dagster.components.core.snapshot import get_package_entry_snap
 
     # create a top-level DefsModule component from the root module
     context = ComponentLoadContext.for_module(defs_root)
-    root_component = DefsModuleComponent.from_context(context)
+    root_component = get_component(context)
     if root_component is None:
         raise DagsterInvalidDefinitionError("Could not resolve root module to a component.")
 

python_modules/dagster/dagster/components/lib/definitions_component/__init__.py

@@ -2,7 +2,6 @@
 from typing import Optional
 
 from dagster._core.definitions.definitions_class import Definitions
-from dagster._core.errors import DagsterInvalidDefinitionError
 from dagster.components.component.component import Component
 from dagster.components.core.context import ComponentLoadContext
 from dagster.components.core.defs_module import DagsterDefsComponent
@@ -20,10 +19,5 @@ class DefinitionsComponent(Component, Model, Resolvable):
     path: Optional[str]
 
     def build_defs(self, context: ComponentLoadContext) -> Definitions:
-        inner_context = context.for_path(Path(self.path)) if self.path else context
-        component = DagsterDefsComponent.from_context(inner_context)
-        if component is None:
-            raise DagsterInvalidDefinitionError(
-                f"Could not resolve {self.path} to a DagsterDefsComponent."
-            )
+        component = DagsterDefsComponent(Path(self.path) if self.path else context.path)
         return component.build_defs(context)

python_modules/libraries/dagster-sling/dagster_sling_tests/test_sling_replication_collection_component.py

@@ -22,7 +22,7 @@
 from dagster._utils.env import environ
 from dagster.components import ComponentLoadContext
 from dagster.components.cli import cli
-from dagster.components.core.defs_module import DefsModuleComponent, WrappedYamlComponent
+from dagster.components.core.defs_module import get_component
 from dagster.components.resolved.context import ResolutionException
 from dagster.components.resolved.core_models import AssetAttributesModel
 from dagster_sling import SlingReplicationCollectionComponent, SlingResource
@@ -80,10 +80,9 @@ def temp_sling_component_instance(
                 data["attributes"]["sling"]["connections"][0]["instance"] = f"{temp_dir}/duckdb"
 
             context = ComponentLoadContext.for_test().for_path(component_path)
-            defs_module = DefsModuleComponent.from_context(context)
-            assert isinstance(defs_module, WrappedYamlComponent)
-            assert isinstance(defs_module.wrapped, SlingReplicationCollectionComponent)
-            yield defs_module.wrapped, defs_module.build_defs(context)
+            component = get_component(context)
+            assert isinstance(component, SlingReplicationCollectionComponent)
+            yield component, component.build_defs(context)
 
 
 def test_python_attributes() -> None: