[components] Enforce that all component schemas are of type ResolvableModel (#27677)

## Summary & Motivation

Doing this has a few benefits. The first is that we force people to use ResolvableModel as the base class for their component schemas, which means we're able to inject additional stuff into that class in the future.

The second is that we can create a default `load()` method, which pretty significantly reduces boilerplate for these component types.

## How I Tested These Changes

## Changelog

NOCHANGELOG

Author: OwenKephart

python_modules/libraries/dagster-components/dagster_components/core/component.py

@@ -16,7 +16,6 @@
 from dagster._core.definitions.definitions_class import Definitions
 from dagster._core.errors import DagsterError
 from dagster._utils import snakecase
-from pydantic import BaseModel
 from typing_extensions import Self
 
 from dagster_components.core.component_key import (
@@ -43,7 +42,7 @@ class Component(ABC):
     name: ClassVar[Optional[str]] = None
 
     @classmethod
-    def get_schema(cls) -> Optional[type[BaseModel]]:
+    def get_schema(cls) -> Optional[type[ResolvableModel]]:
         return None
 
     @classmethod
@@ -65,8 +64,8 @@ def get_additional_scope(cls) -> Mapping[str, Any]:
     def build_defs(self, context: "ComponentLoadContext") -> Definitions: ...
 
     @classmethod
-    @abstractmethod
-    def load(cls, params: Optional[BaseModel], context: "ComponentLoadContext") -> Self: ...
+    def load(cls, params: Optional[ResolvableModel], context: "ComponentLoadContext") -> Self:
+        return cls() if params is None else context.resolve(params, as_type=cls)
 
     @classmethod
     def get_metadata(cls) -> "ComponentTypeInternalMetadata":

python_modules/libraries/dagster-components/dagster_components/lib/dbt_project/component.py

@@ -69,10 +69,6 @@ def get_scaffolder(cls) -> "DbtProjectComponentScaffolder":
     def get_schema(cls) -> type[DbtProjectParams]:
         return DbtProjectParams
 
-    @classmethod
-    def load(cls, params: DbtProjectParams, context: ComponentLoadContext) -> "DbtProjectComponent":
-        return context.resolve(params, as_type=cls)
-
     def get_asset_selection(
         self, select: str, exclude: Optional[str] = None
     ) -> DbtManifestAssetSelection:

python_modules/libraries/dagster-components/dagster_components/lib/definitions_component/component.py

@@ -7,7 +7,6 @@
 )
 from dagster._seven import import_uncached_module_from_path
 from dagster._utils import pushd
-from typing_extensions import Self
 
 from dagster_components import (
     Component,
@@ -26,8 +25,8 @@ class DefinitionsParamSchema(ResolvableModel):
 class DefinitionsComponent(Component):
     """Wraps an arbitrary set of Dagster definitions."""
 
-    def __init__(self, definitions_path: Path):
-        self.definitions_path = definitions_path
+    def __init__(self, definitions_path: Optional[Path]):
+        self.definitions_path = definitions_path or Path("definitions.py")
 
     @classmethod
     def get_scaffolder(cls) -> DefinitionsComponentScaffolder:
@@ -37,10 +36,6 @@ def get_scaffolder(cls) -> DefinitionsComponentScaffolder:
     def get_schema(cls) -> type[DefinitionsParamSchema]:
         return DefinitionsParamSchema
 
-    @classmethod
-    def load(cls, params: DefinitionsParamSchema, context: ComponentLoadContext) -> Self:
-        return cls(definitions_path=Path(params.definitions_path or "definitions.py"))
-
     def build_defs(self, context: ComponentLoadContext) -> Definitions:
         with pushd(str(context.path)):
             module = import_uncached_module_from_path("definitions", str(self.definitions_path))

python_modules/libraries/dagster-components/dagster_components/lib/pipes_subprocess_script_collection.py

@@ -60,12 +60,6 @@ def introspect_from_path(path: Path) -> "PipesSubprocessScriptCollection":
     def get_schema(cls) -> type[PipesSubprocessScriptCollectionParams]:
         return PipesSubprocessScriptCollectionParams
 
-    @classmethod
-    def load(
-        cls, params: PipesSubprocessScriptCollectionParams, context: ComponentLoadContext
-    ) -> "PipesSubprocessScriptCollection":
-        return context.resolve(params, as_type=cls)
-
     def build_defs(self, context: "ComponentLoadContext") -> "Definitions":
         from dagster._core.definitions.definitions_class import Definitions
 

python_modules/libraries/dagster-components/dagster_components/lib/sling_replication_collection/component.py

@@ -9,7 +9,6 @@
 from dagster._record import record
 from dagster_sling import DagsterSlingTranslator, SlingResource, sling_assets
 from dagster_sling.resources import AssetExecutionContext
-from typing_extensions import Self
 
 from dagster_components import Component, ComponentLoadContext
 from dagster_components.core.component import registered_component_type
@@ -112,10 +111,6 @@ def get_scaffolder(cls) -> ComponentScaffolder:
     def get_schema(cls) -> type[SlingReplicationCollectionParams]:
         return SlingReplicationCollectionParams
 
-    @classmethod
-    def load(cls, params: SlingReplicationCollectionParams, context: ComponentLoadContext) -> Self:
-        return context.resolve(params, as_type=cls)
-
     def build_asset(
         self, context: ComponentLoadContext, replication_spec: SlingReplicationSpec
     ) -> AssetsDefinition:

python_modules/libraries/dagster-components/dagster_components/lib/test/all_metadata_empty_asset.py

@@ -1,18 +1,13 @@
 from dagster._core.definitions.decorators.asset_decorator import asset
 from dagster._core.definitions.definitions_class import Definitions
 from dagster._core.execution.context.asset_execution_context import AssetExecutionContext
-from typing_extensions import Self
 
 from dagster_components import Component, ComponentLoadContext, registered_component_type
 from dagster_components.core.component_scaffolder import DefaultComponentScaffolder
 
 
 @registered_component_type(name="all_metadata_empty_asset")
 class AllMetadataEmptyAsset(Component):
-    @classmethod
-    def load(cls, context: "ComponentLoadContext") -> Self:
-        return cls()
-
     @classmethod
     def get_scaffolder(cls) -> DefaultComponentScaffolder:
         return DefaultComponentScaffolder()

python_modules/libraries/dagster-components/dagster_components/lib/test/complex_schema_asset.py

@@ -4,11 +4,10 @@
 from dagster._core.definitions.decorators.asset_decorator import asset
 from dagster._core.definitions.definitions_class import Definitions
 from dagster._core.execution.context.asset_execution_context import AssetExecutionContext
-from pydantic import BaseModel
-from typing_extensions import Self
 
 from dagster_components import Component, ComponentLoadContext, registered_component_type
 from dagster_components.core.component_scaffolder import DefaultComponentScaffolder
+from dagster_components.core.schema.base import ResolvableModel
 from dagster_components.core.schema.metadata import ResolvableFieldInfo
 from dagster_components.core.schema.objects import (
     AssetAttributesModel,
@@ -17,7 +16,7 @@
 )
 
 
-class ComplexAssetParams(BaseModel):
+class ComplexAssetParams(ResolvableModel):
     value: str
     op: Optional[OpSpecModel] = None
     asset_attributes: Annotated[
@@ -38,15 +37,6 @@ def get_schema(cls):
     def get_scaffolder(cls) -> DefaultComponentScaffolder:
         return DefaultComponentScaffolder()
 
-    @classmethod
-    def load(cls, params: ComplexAssetParams, context: "ComponentLoadContext") -> Self:
-        return cls(
-            value=params.value,
-            op_spec=params.op,
-            asset_attributes=params.asset_attributes,
-            asset_transforms=params.asset_transforms or [],
-        )
-
     def __init__(
         self,
         value: str,

python_modules/libraries/dagster-components/dagster_components/lib/test/simple_asset.py

@@ -3,7 +3,6 @@
 from dagster._core.definitions.definitions_class import Definitions
 from dagster._core.execution.context.asset_execution_context import AssetExecutionContext
 from pydantic import BaseModel
-from typing_extensions import Self
 
 from dagster_components import Component, ComponentLoadContext, registered_component_type
 from dagster_components.core.component_scaffolder import (
@@ -29,13 +28,6 @@ def get_schema(cls):
     def get_scaffolder(cls) -> ComponentScaffolder:
         return DefaultComponentScaffolder()
 
-    @classmethod
-    def load(cls, params: SimpleAssetParams, context: "ComponentLoadContext") -> Self:
-        return cls(
-            asset_key=AssetKey.from_user_string(params.asset_key),
-            value=params.value,
-        )
-
     def __init__(self, asset_key: AssetKey, value: str):
         self._asset_key = asset_key
         self._value = value

python_modules/libraries/dagster-components/dagster_components/lib/test/simple_pipes_script_asset.py

@@ -7,7 +7,6 @@
 from dagster._core.execution.context.asset_execution_context import AssetExecutionContext
 from dagster._core.pipes.subprocess import PipesSubprocessClient
 from pydantic import BaseModel
-from typing_extensions import Self
 
 from dagster_components import Component, ComponentLoadContext, registered_component_type
 from dagster_components.core.component_scaffolder import (
@@ -62,13 +61,6 @@ def get_scaffolder(cls) -> ComponentScaffolder:
     def get_schema(cls):
         return SimplePipesScriptAssetParams
 
-    @classmethod
-    def load(cls, params: SimplePipesScriptAssetParams, context: "ComponentLoadContext") -> Self:
-        return cls(
-            asset_key=AssetKey.from_user_string(params.asset_key),
-            script_path=context.path / params.filename,
-        )
-
     def __init__(self, asset_key: AssetKey, script_path: Path):
         self._asset_key = asset_key
         self._script_path = script_path

python_modules/libraries/dagster-components/dagster_components/test/basic_components.py

@@ -4,13 +4,12 @@
 
 from dagster._core.definitions.definitions_class import Definitions
 from pydantic import BaseModel, ConfigDict
-from typing_extensions import Self
 
-from dagster_components import Component, registered_component_type
+from dagster_components import Component, ResolvableModel, registered_component_type
 from dagster_components.core.component import ComponentLoadContext
 
 
-class MyComponentSchema(BaseModel):
+class MyComponentSchema(ResolvableModel):
     a_string: str
     an_int: int
 
@@ -21,14 +20,14 @@ class MyComponentSchema(BaseModel):
 class MyComponent(Component):
     name = "my_component"
 
+    def __init__(self, a_string: str, an_int: int):
+        self.a_string = a_string
+        self.an_int = an_int
+
     @classmethod
     def get_schema(cls) -> type[MyComponentSchema]:
         return MyComponentSchema
 
-    @classmethod
-    def load(cls, params: MyComponentSchema, context: ComponentLoadContext) -> Self:
-        return cls()
-
     def build_defs(self, context: ComponentLoadContext) -> Definitions:
         return Definitions()
 
@@ -54,9 +53,5 @@ class MyNestedComponent(Component):
     def get_schema(cls) -> type[MyNestedComponentSchema]:
         return MyNestedComponentSchema
 
-    @classmethod
-    def load(cls, params: MyComponentSchema, context: ComponentLoadContext) -> Self:
-        return cls()
-
     def build_defs(self, context: ComponentLoadContext) -> Definitions:
         return Definitions()

python_modules/libraries/dagster-components/dagster_components_tests/integration_tests/components/definitions/local_component_sample/__init__.py

@@ -1,11 +1,9 @@
 from dagster._core.definitions.definitions_class import Definitions
-from dagster_components import Component, registered_component_type
+from dagster_components import Component, ResolvableModel, registered_component_type
 from dagster_components.core.component import ComponentLoadContext
-from dagster_components.core.schema.base import BaseModel
-from typing_extensions import Self
 
 
-class MyComponentSchema(BaseModel):
+class MyComponentSchema(ResolvableModel):
     a_string: str
     an_int: int
 
@@ -14,13 +12,13 @@ class MyComponentSchema(BaseModel):
 class MyComponent(Component):
     name = "my_component"
 
+    def __init__(self, a_string: str, an_int: int):
+        self.a_string = a_string
+        self.an_int = an_int
+
     @classmethod
     def get_schema(cls):
         return MyComponentSchema
 
-    @classmethod
-    def load(cls, params: MyComponentSchema, context: ComponentLoadContext) -> Self:
-        return cls()
-
     def build_defs(self, context: ComponentLoadContext) -> Definitions:
         return Definitions()

python_modules/libraries/dagster-components/dagster_components_tests/integration_tests/components/definitions/other_local_component_sample/__init__.py

@@ -2,7 +2,6 @@
 from dagster_components import Component, registered_component_type
 from dagster_components.core.component import ComponentLoadContext
 from pydantic import BaseModel
-from typing_extensions import Self
 
 
 class MyNewComponentSchema(BaseModel):
@@ -18,9 +17,5 @@ class MyNewComponent(Component):
     def get_schema(cls):
         return MyNewComponentSchema
 
-    @classmethod
-    def load(cls, params: MyNewComponentSchema, context: ComponentLoadContext) -> Self:
-        return cls()
-
     def build_defs(self, context: ComponentLoadContext) -> Definitions:
         return Definitions()

python_modules/libraries/dagster-dg/dagster_dg/templates/COMPONENT_TYPE/COMPONENT_TYPE_NAME_PLACEHOLDER.py.jinja

@@ -17,6 +17,8 @@ class {{ component_type_class_name }}(Component):
     COMPONENT DESCRIPTION HERE.
     """
 
+    def __init__(self): ...
+
     @classmethod
     def get_schema(cls):
         return {{ component_type_class_name }}Params
@@ -25,15 +27,6 @@ class {{ component_type_class_name }}(Component):
     def get_scaffolder(cls) -> DefaultComponentScaffolder:
         return DefaultComponentScaffolder()
 
-    @classmethod
-    def load(
-        cls,
-        params: {{ component_type_class_name }}Params,
-        context: ComponentLoadContext,
-    ) -> "{{ component_type_class_name }}":
-        # Add logic for mapping schema parameters to constructor args here.
-        return cls()
-
     def build_defs(self, load_context: ComponentLoadContext) -> Definitions:
         # Add definition construction logic here.
         return Definitions()