[components] Use AssetAttributesModel instead of DbtTranslatorParams

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

@@ -80,19 +80,27 @@ class RenderedModel(BaseModel):
 
     model_config = ConfigDict(json_schema_extra={JSON_SCHEMA_EXTRA_KEY: True})
 
+    def _render_property(
+        self, key: str, raw_value: Any, value_resolver: "TemplatedValueResolver"
+    ) -> Any:
+        return value_resolver.render_obj(raw_value)
+
     def render_properties(self, value_resolver: "TemplatedValueResolver") -> Mapping[str, Any]:
         """Returns a dictionary of rendered properties for this class."""
-        rendered_properties = value_resolver.render_obj(self.model_dump(exclude_unset=True))
+        raw_properties = self.model_dump(exclude_unset=True)
 
         # validate that the rendered properties match the output type
-        for k, v in rendered_properties.items():
+        rendered_properties = {}
+        for k, v in raw_properties.items():
+            rendered = self._render_property(k, v, value_resolver)
             annotation = self.__annotations__[k]
             expected_type = _get_expected_type(annotation)
             if expected_type is not None:
                 # hook into pydantic's type validation to handle complicated stuff like Optional[Mapping[str, int]]
                 TypeAdapter(
                     expected_type, config={"arbitrary_types_allowed": True}
-                ).validate_python(v)
+                ).validate_python(rendered)
+            rendered_properties[k] = rendered
 
         return rendered_properties
 

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

@@ -25,7 +25,7 @@ class OpSpecBaseModel(BaseModel):
 
 
 class AssetAttributesModel(RenderedModel):
-    key: Optional[str] = None
+    key: Annotated[Optional[str], RenderingMetadata(output_type=AssetKey)] = None
     deps: Sequence[str] = []
     description: Optional[str] = None
     metadata: Annotated[
@@ -42,6 +42,13 @@ class AssetAttributesModel(RenderedModel):
         Optional[str], RenderingMetadata(output_type=Optional[AutomationCondition])
     ] = None
 
+    def _render_property(self, key, raw_value, value_resolver):
+        rendered = super()._render_property(key, raw_value, value_resolver)
+        if key == "key":
+            # coerce the string asset key into an AssetKey object
+            return AssetKey.from_user_string(rendered) if rendered else None
+        return rendered
+
 
 class AssetSpecProcessor(ABC, BaseModel):
     target: str = "*"

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

@@ -17,20 +17,19 @@
     TemplatedValueResolver,
     component_type,
 )
-from dagster_components.core.component_rendering import RenderedModel
-from dagster_components.core.dsl_schema import AssetAttributes, AssetSpecProcessor, OpSpecBaseModel
+from dagster_components.core.dsl_schema import (
+    AssetAttributes,
+    AssetAttributesModel,
+    AssetSpecProcessor,
+    OpSpecBaseModel,
+)
 from dagster_components.generate import generate_component_yaml
 
 
-class DbtNodeTranslatorParams(RenderedModel):
-    key: Optional[str] = None
-    group: Optional[str] = None
-
-
 class DbtProjectParams(BaseModel):
     dbt: DbtCliResource
     op: Optional[OpSpecBaseModel] = None
-    translator: Optional[DbtNodeTranslatorParams] = None
+    translator: Optional[AssetAttributesModel] = None
     asset_attributes: Optional[AssetAttributes] = None
 
 
@@ -43,28 +42,37 @@ class DbtProjectComponentTranslator(DagsterDbtTranslator):
     def __init__(
         self,
         *,
+        params: Optional[AssetAttributesModel],
         value_resolver: TemplatedValueResolver,
-        translator_params: Optional[DbtNodeTranslatorParams] = None,
     ):
+        self.params = params or AssetAttributesModel()
         self.value_resolver = value_resolver
-        self.translator_params = translator_params
+
+    def _get_rendered_attribute(
+        self, attribute: str, dbt_resource_props: Mapping[str, Any], default_method
+    ) -> Any:
+        resolver = self.value_resolver.with_context(node=dbt_resource_props)
+        rendered_attribute = self.params.render_properties(resolver).get(attribute)
+        return (
+            rendered_attribute
+            if rendered_attribute is not None
+            else default_method(dbt_resource_props)
+        )
 
     def get_asset_key(self, dbt_resource_props: Mapping[str, Any]) -> AssetKey:
-        if not self.translator_params or not self.translator_params.key:
-            return super().get_asset_key(dbt_resource_props)
+        return self._get_rendered_attribute("key", dbt_resource_props, super().get_asset_key)
 
-        return AssetKey.from_user_string(
-            self.value_resolver.with_context(node=dbt_resource_props).render_obj(
-                self.translator_params.key
-            )
+    def get_group_name(self, dbt_resource_props: Mapping[str, Any]) -> Optional[str]:
+        return self._get_rendered_attribute(
+            "group_name", dbt_resource_props, super().get_group_name
         )
 
-    def get_group_name(self, dbt_resource_props) -> Optional[str]:
-        if not self.translator_params or not self.translator_params.group:
-            return super().get_group_name(dbt_resource_props)
+    def get_tags(self, dbt_resource_props):
+        return self._get_rendered_attribute("tags", dbt_resource_props, super().get_tags)
 
-        return self.value_resolver.with_context(node=dbt_resource_props).render_obj(
-            self.translator_params.group
+    def get_automation_condition(self, dbt_resource_props):
+        return self._get_rendered_attribute(
+            "automation_condition", dbt_resource_props, super().get_automation_condition
         )
 
 
@@ -93,7 +101,7 @@ def load(cls, context: ComponentLoadContext) -> Self:
             dbt_resource=loaded_params.dbt,
             op_spec=loaded_params.op,
             dbt_translator=DbtProjectComponentTranslator(
-                translator_params=loaded_params.translator,
+                params=loaded_params.translator,
                 value_resolver=context.templated_value_resolver,
             ),
             asset_processors=loaded_params.asset_attributes or [],

python_modules/libraries/dagster-components/dagster_components_tests/integration_tests/test_templated_custom_keys_dbt_project.py

@@ -87,7 +87,7 @@ def test_python_params_group(dbt_path: Path) -> None:
             params={
                 "dbt": {"project_dir": "jaffle_shop"},
                 "translator": {
-                    "group": "some_group",
+                    "group_name": "some_group",
                 },
             },
         ),
@@ -126,7 +126,7 @@ def test_render_vars_root(dbt_path: Path) -> None:
                 params={
                     "dbt": {"project_dir": "jaffle_shop"},
                     "translator": {
-                        "group": "{{ env('GROUP_AS_ENV') }}",
+                        "group_name": "{{ env('GROUP_AS_ENV') }}",
                     },
                 },
             ),