[components][rfc] Rename "ResolvableModel" to "ComponentSchema" (#27678)

## Summary & Motivation

If we're going to enforce that this is always used with Components, we should probably name it something more specific.

## How I Tested These Changes

## Changelog

NOCHANGELOG

Author: OwenKephart

docs/docs/guides/preview/components/creating-a-component.md

@@ -62,13 +62,15 @@ In this case, we'll want to define a few things:
 
 To simplify common use cases, `dagster-components` provides schemas for common bits of configuration:
 
-- `AssetAttributesModel`: This contains attributes that are common to all assets, such as the key, description, tags, and dependencies.
-- `OpSpecModel`: This contains attributes specific to an underlying operation, such as the name and tags.
+- `AssetSpecSchema`: This contains attributes that are common to all assets, such as the key, description, tags, and dependencies.
+- `OpSpecSchema`: This contains attributes specific to an underlying operation, such as the name and tags.
 
 We can the schema for our component and add it to our class as follows:
 
 <CodeExample path="docs_beta_snippets/docs_beta_snippets/guides/components/shell-script-component/with-config-schema.py" language="python" />
 
+Because the argument names in the schema match the names of the arguments in the `ShellCommandComponent` class, the `load` method will automatically populate the class with the values from the schema, and will automatically resolve the `AssetSpecSchema`s into `AssetSpec` objects.
+
 ## Building definitions
 
 Now that we've defined how the component is parameterized, we need to define how to turn those parameters into a `Definitions` object.

examples/docs_beta_snippets/docs_beta_snippets/guides/components/shell-script-component/defining-resolvable-field.py

@@ -1,16 +1,16 @@
 from typing import Annotated, Optional
 
 from dagster_components import ResolvableFieldInfo
-from dagster_components.core.schema.objects import AssetAttributesModel, OpSpecModel
+from dagster_components.core.schema.objects import AssetAttributesSchema, OpSpecSchema
 from pydantic import BaseModel
 
 
 class ShellScriptSchema(BaseModel):
     script_path: str
-    asset_attributes: AssetAttributesModel
+    asset_attributes: AssetAttributesSchema
     # highlight-start
     script_runner: Annotated[
         str, ResolvableFieldInfo(required_scope={"get_script_runner"})
     ]
     # highlight-end
-    op: Optional[OpSpecModel] = None
+    op: Optional[OpSpecSchema] = None

examples/docs_beta_snippets/docs_beta_snippets/guides/components/shell-script-component/defining-resolver.py

@@ -0,0 +1,17 @@
+from collections.abc import Sequence
+from typing import Annotated, Optional
+
+from dagster_components import ComponentSchema, ResolutionContext, Resolver, resolver
+from dagster_components.core.schema.objects import AssetAttributesSchema, OpSpecSchema
+from pydantic import BaseModel
+
+
+class ShellScriptSchema(ComponentSchema):
+    script_path: str
+    asset_attributes: Sequence[AssetAttributesSchema]
+    op: Optional[OpSpecSchema] = None
+
+
+@resolver(fromtype=ShellScriptSchema)
+class ShellScriptResolver(Resolver[ShellScriptSchema]):
+    def resolve_my_object(self, context: ResolutionContext): ...

examples/docs_beta_snippets/docs_beta_snippets/guides/components/shell-script-component/resolving-resolvable-field.py

@@ -1,34 +1,17 @@
+from collections.abc import Sequence
+
 from dagster_components import (
+    AssetSpecSchema,
     Component,
-    ComponentLoadContext,
+    ComponentSchema,
     registered_component_type,
 )
 
-import dagster as dg
-
-
-class ScriptRunner: ...
 
-
-def _get_script_runner(val: str) -> ScriptRunner:
-    return ScriptRunner()
+class ShellCommandParams(ComponentSchema):
+    path: str
+    asset_specs: Sequence[AssetSpecSchema]
 
 
 @registered_component_type(name="shell_command")
-class ShellCommand(Component):
-    def __init__(self, params):
-        self.params = params
-
-    ...
-
-    def build_defs(self, load_context: ComponentLoadContext) -> dg.Definitions:
-        # highlight-start
-        # resolve the script runner with its required additional scope
-        script_runner = self.params.resolve_properties(
-            load_context.resolution_context.with_scope(
-                get_script_runner=_get_script_runner
-            )
-        )["script_runner"]
-        # highlight-end
-        ...
-        return dg.Definitions(...)
+class ShellCommand(Component): ...

examples/docs_beta_snippets/docs_beta_snippets/guides/components/shell-script-component/with-build-defs.py

@@ -1,59 +1,47 @@
 import subprocess
+from collections.abc import Sequence
 from typing import Optional
 
 from dagster_components import (
+    AssetSpecSchema,
     Component,
     ComponentLoadContext,
+    ComponentSchema,
+    OpSpecSchema,
     registered_component_type,
 )
-from dagster_components.core.schema.objects import AssetAttributesModel, OpSpecModel
-from pydantic import BaseModel
 
 import dagster as dg
 
 
-# highlight-start
-class ShellScriptSchema(BaseModel):
+class ShellScriptSchema(ComponentSchema):
     script_path: str
-    asset_attributes: AssetAttributesModel
-    op: Optional[OpSpecModel] = None
-    # highlight-end
+    asset_specs: Sequence[AssetSpecSchema]
+    op: Optional[OpSpecSchema] = None
 
 
 @registered_component_type(name="shell_command")
 class ShellCommand(Component):
-    def __init__(self, params: ShellScriptSchema):
-        self.params = params
+    def __init__(
+        self,
+        script_path: str,
+        asset_specs: Sequence[dg.AssetSpec],
+        op: Optional[OpSpecSchema] = None,
+    ):
+        self.script_path = script_path
+        self.specs = asset_specs
+        self.op = op or OpSpecSchema()
 
     @classmethod
     def get_schema(cls) -> type[ShellScriptSchema]:
-        # higlight-start
         return ShellScriptSchema
-        # highlight-end
-
-    @classmethod
-    def load(
-        cls, params: ShellScriptSchema, load_context: ComponentLoadContext
-    ) -> "ShellCommand":
-        return cls(params=params)
 
     def build_defs(self, load_context: ComponentLoadContext) -> dg.Definitions:
-        resolved_asset_attributes = (
-            self.params.asset_attributes.get_resolved_properties(
-                load_context.resolution_context
-            )
-        )
-        resolved_op_properties = (
-            self.params.op.get_resolved_properties(load_context.resolution_context)
-            if self.params.op
-            else {}
-        )
-
-        @dg.asset(**resolved_asset_attributes, **resolved_op_properties)
+        @dg.multi_asset(specs=self.specs, op_tags=self.op.tags, name=self.op.name)
         def _asset(context: dg.AssetExecutionContext):
             self.execute(context)
 
         return dg.Definitions(assets=[_asset])
 
     def execute(self, context: dg.AssetExecutionContext):
-        subprocess.run(["sh", self.params.script_path], check=False)
+        subprocess.run(["sh", self.script_path], check=True)

examples/docs_beta_snippets/docs_beta_snippets/guides/components/shell-script-component/with-config-schema.py

@@ -1,30 +1,39 @@
+from collections.abc import Sequence
 from typing import Optional
 
 from dagster_components import (
+    AssetSpecSchema,
     Component,
     ComponentLoadContext,
+    ComponentSchema,
+    OpSpecSchema,
     registered_component_type,
 )
-from dagster_components.core.schema.objects import AssetAttributesModel, OpSpecModel
 from pydantic import BaseModel
 
-from dagster import Definitions
+import dagster as dg
 
 
-# highlight-start
-class ShellScriptSchema(BaseModel):
+class ShellScriptSchema(ComponentSchema):
     script_path: str
-    asset_attributes: AssetAttributesModel
-    op: Optional[OpSpecModel] = None
-    # highlight-end
+    asset_specs: Sequence[AssetSpecSchema]
+    op: Optional[OpSpecSchema] = None
 
 
 @registered_component_type(name="shell_command")
 class ShellCommand(Component):
+    def __init__(
+        self,
+        script_path: str,
+        asset_specs: Sequence[dg.AssetSpec],
+        op: Optional[OpSpecSchema] = None,
+    ):
+        self.script_path = script_path
+        self.specs = asset_specs
+        self.op = op or OpSpecSchema()
+
     @classmethod
     def get_schema(cls) -> type[ShellScriptSchema]:
-        # higlight-start
         return ShellScriptSchema
-        # highlight-end
 
-    def build_defs(self, load_context: ComponentLoadContext) -> Definitions: ...
+    def build_defs(self, load_context: ComponentLoadContext) -> dg.Definitions: ...

examples/experimental/dagster-blueprints/dagster_blueprints/blueprint_assets_definition.py

@@ -13,7 +13,7 @@
 from dagster_blueprints.blueprint import Blueprint
 
 
-class AssetSpecModel(DagsterModel):
+class AssetSpecSchema(DagsterModel):
     key: str
     deps: Sequence[str] = []
     description: Optional[str] = None
@@ -36,7 +36,7 @@ def to_asset_spec(self) -> AssetSpec:
 class BlueprintAssetsDefinition(Blueprint):
     """A blueprint that produces an AssetsDefinition."""
 
-    assets: Sequence[AssetSpecModel]
+    assets: Sequence[AssetSpecSchema]
 
     def build_defs(self) -> Definitions:
         specs = [spec_model.to_asset_spec() for spec_model in self.assets]

examples/experimental/dagster-blueprints/dagster_blueprints_tests/test_databricks_blueprint.py

@@ -10,7 +10,7 @@
     materialize,
 )
 from dagster._core.pipes.client import PipesClientCompletedInvocation
-from dagster_blueprints.blueprint_assets_definition import AssetSpecModel
+from dagster_blueprints.blueprint_assets_definition import AssetSpecSchema
 from dagster_blueprints.databricks_blueprint import DatabricksTaskBlueprint
 from databricks.sdk.service import jobs
 
@@ -57,7 +57,7 @@ def run(self, *, context, extras=None, **kwargs) -> PipesClientCompletedInvocati
 def test_single_databricks_task_blueprint() -> None:
     databricks_task_dict = make_submit_task_dict("/my/script/path.py", "/my/whl/path.whl", True)
     single_asset_blueprint = DatabricksTaskBlueprint(
-        assets=[AssetSpecModel(key="asset1")], task=databricks_task_dict
+        assets=[AssetSpecSchema(key="asset1")], task=databricks_task_dict
     )
     defs = single_asset_blueprint.build_defs()
     asset1 = cast(AssetsDefinition, next(iter(defs.assets or [])))
@@ -76,7 +76,7 @@ def test_single_databricks_task_blueprint() -> None:
 def test_single_databricks_task_blueprint_with_result() -> None:
     databricks_task_dict = make_submit_task_dict("/my/script/path.py", "/my/whl/path.whl", True)
     single_asset_blueprint = DatabricksTaskBlueprint(
-        assets=[AssetSpecModel(key="asset1")], task=databricks_task_dict
+        assets=[AssetSpecSchema(key="asset1")], task=databricks_task_dict
     )
     defs = single_asset_blueprint.build_defs()
     asset1 = cast(AssetsDefinition, next(iter(defs.assets or [])))
@@ -98,7 +98,7 @@ def test_single_databricks_task_blueprint_with_result() -> None:
 def test_multi_asset_databricks_task_blueprint() -> None:
     databricks_task_dict = make_submit_task_dict("/my/script/path.py", "/my/whl/path.whl", True)
     multi_asset_blueprint = DatabricksTaskBlueprint(
-        assets=[AssetSpecModel(key="asset1"), AssetSpecModel(key="asset2")],
+        assets=[AssetSpecSchema(key="asset1"), AssetSpecSchema(key="asset2")],
         task=databricks_task_dict,
     )
     defs = multi_asset_blueprint.build_defs()
@@ -118,7 +118,7 @@ def test_multi_asset_databricks_task_blueprint() -> None:
 def test_multi_asset_databricks_task_blueprint_with_results() -> None:
     databricks_task_dict = make_submit_task_dict("/my/script/path.py", "/my/whl/path.whl", True)
     multi_asset_blueprint = DatabricksTaskBlueprint(
-        assets=[AssetSpecModel(key="asset1"), AssetSpecModel(key="asset2")],
+        assets=[AssetSpecSchema(key="asset1"), AssetSpecSchema(key="asset2")],
         task=databricks_task_dict,
     )
     defs = multi_asset_blueprint.build_defs()
@@ -149,10 +149,10 @@ def test_multi_asset_databricks_task_blueprint_with_results() -> None:
 
 def test_op_name_collisions() -> None:
     single_asset_blueprint1 = DatabricksTaskBlueprint(
-        assets=[AssetSpecModel(key="asset1")], task={"placeholder": "placeholder"}
+        assets=[AssetSpecSchema(key="asset1")], task={"placeholder": "placeholder"}
     )
     single_asset_blueprint2 = DatabricksTaskBlueprint(
-        assets=[AssetSpecModel(key="asset2")], task={"placeholder": "placeholder"}
+        assets=[AssetSpecSchema(key="asset2")], task={"placeholder": "placeholder"}
     )
     resources = {"pipes_databricks_client": object()}
     blueprint_defs = Definitions.merge(

examples/experimental/dagster-blueprints/dagster_blueprints_tests/test_shell_command_blueprint.py

@@ -13,7 +13,7 @@
     DATA_VERSION_TAG,
 )
 from dagster._core.pipes.subprocess import PipesSubprocessClient
-from dagster_blueprints.blueprint_assets_definition import AssetSpecModel
+from dagster_blueprints.blueprint_assets_definition import AssetSpecSchema
 from dagster_blueprints.shell_command_blueprint import ShellCommandBlueprint
 
 
@@ -29,7 +29,7 @@ def temp_script(script_fn: Callable[[], Any]) -> Iterator[str]:
 
 def test_single_asset_shell_command_blueprint() -> None:
     single_asset_blueprint = ShellCommandBlueprint(
-        assets=[AssetSpecModel(key="asset1")], command=["echo", '"hello"']
+        assets=[AssetSpecSchema(key="asset1")], command=["echo", '"hello"']
     )
     defs = single_asset_blueprint.build_defs()
     asset1 = cast(AssetsDefinition, next(iter(defs.assets or [])))
@@ -41,7 +41,7 @@ def test_single_asset_shell_command_blueprint() -> None:
 
 def test_single_asset_shell_command_blueprint_key_prefix() -> None:
     single_asset_blueprint = ShellCommandBlueprint(
-        assets=[AssetSpecModel(key="prefix/asset1")], command=["echo", '"hello"']
+        assets=[AssetSpecSchema(key="prefix/asset1")], command=["echo", '"hello"']
     )
     defs = single_asset_blueprint.build_defs()
     asset1 = cast(AssetsDefinition, next(iter(defs.assets or [])))
@@ -50,7 +50,7 @@ def test_single_asset_shell_command_blueprint_key_prefix() -> None:
 
 def test_single_asset_shell_command_blueprint_str_command() -> None:
     single_asset_blueprint = ShellCommandBlueprint(
-        assets=[AssetSpecModel(key="asset1")], command='echo "hello world"'
+        assets=[AssetSpecSchema(key="asset1")], command='echo "hello world"'
     )
     defs = single_asset_blueprint.build_defs()
     asset1 = cast(AssetsDefinition, next(iter(defs.assets or [])))
@@ -74,7 +74,7 @@ def script_fn():
     extras = {"bar": "baz"}
     with temp_script(script_fn) as script_path:
         single_asset_blueprint = ShellCommandBlueprint(
-            assets=[AssetSpecModel(key="asset1")],
+            assets=[AssetSpecSchema(key="asset1")],
             command=[cast(str, shutil.which("python")), script_path],
             extras=extras,
         )
@@ -99,7 +99,7 @@ def script_fn():
 
 def test_multi_asset_shell_command_blueprint() -> None:
     multi_asset_blueprint = ShellCommandBlueprint(
-        assets=[AssetSpecModel(key="asset1"), AssetSpecModel(key="asset2")],
+        assets=[AssetSpecSchema(key="asset1"), AssetSpecSchema(key="asset2")],
         command=["echo", '"hello"'],
     )
     defs = multi_asset_blueprint.build_defs()
@@ -121,7 +121,7 @@ def script_fn():
 
     with temp_script(script_fn) as script_path:
         multi_asset_blueprint = ShellCommandBlueprint(
-            assets=[AssetSpecModel(key="asset1"), AssetSpecModel(key="asset2")],
+            assets=[AssetSpecSchema(key="asset1"), AssetSpecSchema(key="asset2")],
             command=[cast(str, shutil.which("python")), script_path],
         )
 
@@ -146,10 +146,10 @@ def script_fn():
 
 def test_op_name_collisions() -> None:
     single_asset_blueprint1 = ShellCommandBlueprint(
-        assets=[AssetSpecModel(key="asset1")], command=["echo", '"hello"']
+        assets=[AssetSpecSchema(key="asset1")], command=["echo", '"hello"']
     )
     single_asset_blueprint2 = ShellCommandBlueprint(
-        assets=[AssetSpecModel(key="asset2")], command=["echo", '"hello"']
+        assets=[AssetSpecSchema(key="asset2")], command=["echo", '"hello"']
     )
     resources = {"pipes_subprocess_client": PipesSubprocessClient()}
     blueprint_defs = Definitions.merge(

python_modules/libraries/dagster-components/dagster_components/__init__.py

@@ -13,12 +13,15 @@
     ComponentScaffoldRequest as ComponentScaffoldRequest,
     DefaultComponentScaffolder as DefaultComponentScaffolder,
 )
-from dagster_components.core.schema.base import ResolvableModel as ResolvableModel
+from dagster_components.core.schema.base import ComponentSchema as ComponentSchema
 from dagster_components.core.schema.context import ResolutionContext as ResolutionContext
 from dagster_components.core.schema.metadata import ResolvableFieldInfo as ResolvableFieldInfo
 from dagster_components.core.schema.objects import (
-    AssetAttributesModel as AssetAttributesModel,
-    AssetSpecTransformModel as AssetSpecTransformModel,
-    OpSpecModel as OpSpecModel,
+    AssetAttributesSchema as AssetAttributesSchema,
+    AssetSpecSchema as AssetSpecSchema,
+    AssetSpecTransformSchema as AssetSpecTransformSchema,
+    OpSpecSchema as OpSpecSchema,
+    Resolver as Resolver,
+    resolver as resolver,
 )
 from dagster_components.version import __version__ as __version__