Update callsites of AssetSpec._replace to use replace_attributes

docs/content/integrations/looker.mdx

@@ -112,11 +112,13 @@ class CustomDagsterLookerApiTranslator(DagsterLookerApiTranslator):
         asset_spec = super().get_asset_spec(looker_structure)
 
         # Add a team owner tag for all Looker assets
-        asset_spec = asset_spec._replace(owners=["team:my_team"])
+        asset_spec = asset_spec.replace_attributes(owners=["team:my_team"])
 
         # For only Looker dashboard, prefix the asset key with "looker" for organizational purposes
         if looker_structure.structure_type == LookerStructureType.DASHBOARD:
-            asset_spec = asset_spec._replace(key=asset_spec.key.with_prefix("looker"))
+            asset_spec = asset_spec.replace_attributes(
+                key=asset_spec.key.with_prefix("looker")
+            )
 
         return asset_spec
 

examples/docs_snippets/docs_snippets/integrations/looker/customize-looker-assets.py

@@ -20,11 +20,13 @@ def get_asset_spec(self, looker_structure: LookerStructureData) -> dg.AssetSpec:
         asset_spec = super().get_asset_spec(looker_structure)
 
         # Add a team owner tag for all Looker assets
-        asset_spec = asset_spec._replace(owners=["team:my_team"])
+        asset_spec = asset_spec.replace_attributes(owners=["team:my_team"])
 
         # For only Looker dashboard, prefix the asset key with "looker" for organizational purposes
         if looker_structure.structure_type == LookerStructureType.DASHBOARD:
-            asset_spec = asset_spec._replace(key=asset_spec.key.with_prefix("looker"))
+            asset_spec = asset_spec.replace_attributes(
+                key=asset_spec.key.with_prefix("looker")
+            )
 
         return asset_spec
 

examples/experimental/dagster-airlift/dagster_airlift/core/serialization/defs_construction.py

@@ -45,10 +45,9 @@ def enrich_spec_with_airflow_task_metadata(
     tasks: AbstractSet[TaskHandle],
     serialized_data: SerializedAirflowDefinitionsData,
 ) -> AssetSpec:
-    tags = {**spec.tags, **airlift_mapped_kind_dict()}
-    return spec._replace(
-        metadata={**spec.metadata, **metadata_for_mapped_tasks(tasks, serialized_data)},
-        tags=tags,
+    return spec.merge_attributes(
+        metadata=metadata_for_mapped_tasks(tasks, serialized_data),
+        tags=airlift_mapped_kind_dict(),
     )
 
 
@@ -65,10 +64,9 @@ def enrich_spec_with_airflow_dag_metadata(
     dags: AbstractSet[DagHandle],
     serialized_data: SerializedAirflowDefinitionsData,
 ) -> AssetSpec:
-    tags = {**spec.tags, **airlift_mapped_kind_dict()}
-    return spec._replace(
-        metadata={**spec.metadata, **metadata_for_mapped_dags(dags, serialized_data)},
-        tags=tags,
+    return spec.merge_attributes(
+        metadata=metadata_for_mapped_dags(dags, serialized_data),
+        tags=airlift_mapped_kind_dict(),
     )
 
 

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

@@ -67,7 +67,7 @@ def _attach_code_references_to_definitions(blueprint: Blueprint, defs: Definitio
 
         new_assets_defs.append(
             assets_def.map_asset_specs(
-                lambda spec: spec._replace(metadata=new_metadata_by_key[spec.key])
+                lambda spec: spec.replace_attributes(metadata=new_metadata_by_key[spec.key])
             )
         )
     return copy(defs, assets=new_assets_defs)

examples/starlift-demo/dbt_example/dagster_defs/dbt_cloud_utils.py

@@ -52,7 +52,7 @@ def filter_specs_by_tag(specs: Sequence[AssetSpec], tag: str) -> Dict[AssetKey,
 
 
 def add_dep_to_spec(spec: AssetSpec, dep: AssetKey) -> AssetSpec:
-    return spec._replace(deps=[*spec.deps, AssetDep(dep)])
+    return spec.replace_attributes(deps=[*spec.deps, AssetDep(dep)])
 
 
 def key_for_uid(specs: Sequence[AssetSpec], uid: str) -> AssetKey:

examples/starlift-demo/dbt_example/dagster_defs/jaffle_shop.py

@@ -24,7 +24,7 @@ def jaffle_shop_assets(context: AssetExecutionContext, dbt: DbtCliResource):
 
 
 jaffle_shop_external_assets = [
-    spec._replace(code_version=None, skippable=False) for spec in jaffle_shop_assets.specs
+    spec.replace_attributes(code_version=None, skippable=False) for spec in jaffle_shop_assets.specs
 ]
 
 jaffle_shop_with_upstream = eager_asset(with_deps(DBT_SOURCE_TO_DAG, jaffle_shop_assets))

examples/starlift-demo/dbt_example/dagster_defs/utils.py

@@ -25,7 +25,9 @@ def apply_eager_automation(defs: Definitions) -> Definitions:
             continue
         assets.append(
             asset.map_asset_specs(
-                lambda spec: spec._replace(automation_condition=AutomationCondition.eager())
+                lambda spec: spec.replace_attributes(
+                    automation_condition=AutomationCondition.eager()
+                )
                 if spec.automation_condition is None
                 else spec
             )
@@ -40,7 +42,7 @@ def apply_eager_automation(defs: Definitions) -> Definitions:
 
 
 def with_group(assets_def: AssetsDefinition, group_name: str) -> AssetsDefinition:
-    return assets_def.map_asset_specs(lambda spec: spec._replace(group_name=group_name))
+    return assets_def.map_asset_specs(lambda spec: spec.replace_attributes(group_name=group_name))
 
 
 def with_deps(
@@ -93,4 +95,6 @@ def with_deps(
 
 
 def eager(specs: Sequence[AssetSpec]) -> Sequence[AssetSpec]:
-    return [spec._replace(automation_condition=AutomationCondition.eager()) for spec in specs]
+    return [
+        spec.replace_attributes(automation_condition=AutomationCondition.eager()) for spec in specs
+    ]

python_modules/dagster/dagster/_core/definitions/assets.py

@@ -1238,7 +1238,7 @@ def update_replace_dict_and_conflicts(
 
                 replace_dict["deps"] = new_deps
 
-            replaced_specs.append(spec._replace(**replace_dict))
+            replaced_specs.append(spec.replace_attributes(**replace_dict))
 
         for attr_name, conflicting_asset_keys in conflicts_by_attr_name.items():
             raise DagsterInvalidDefinitionError(

python_modules/dagster/dagster/_core/definitions/decorators/decorator_assets_definition_builder.py

@@ -588,7 +588,7 @@ def specs(self) -> Sequence[AssetSpec]:
             " AssetSpecs/AssetOuts supplied to this multi_asset have a group_name defined.",
         )
 
-        return [spec._replace(group_name=self.group_name) for spec in specs]
+        return [spec.replace_attributes(group_name=self.group_name) for spec in specs]
 
     def _synthesize_specs(self) -> Sequence[AssetSpec]:
         resolved_specs = []

python_modules/dagster/dagster/_core/definitions/external_asset.py

@@ -183,19 +183,17 @@ def create_unexecutable_external_asset_from_assets_def(
             for key in assets_def.keys:
                 orig_spec = assets_def.get_asset_spec(key)
                 specs.append(
-                    orig_spec._replace(
-                        metadata={
-                            **(orig_spec.metadata or {}),
-                            **(
-                                {
-                                    SYSTEM_METADATA_KEY_IO_MANAGER_KEY: assets_def.get_io_manager_key_for_asset_key(
-                                        key
-                                    )
-                                }
-                                if assets_def.has_output_for_asset_key(key)
-                                else {}
-                            ),
-                        },
+                    orig_spec.merge_attributes(
+                        metadata=(
+                            {
+                                SYSTEM_METADATA_KEY_IO_MANAGER_KEY: assets_def.get_io_manager_key_for_asset_key(
+                                    key
+                                )
+                            }
+                            if assets_def.has_output_for_asset_key(key)
+                            else {}
+                        ),
+                    ).replace_attributes(
                         automation_condition=None,
                         freshness_policy=None,
                     )

python_modules/dagster/dagster/_core/definitions/metadata/source_code.py

@@ -144,7 +144,7 @@ def _with_code_source_single_definition(
             }
 
     return assets_def.map_asset_specs(
-        lambda spec: spec._replace(metadata=metadata_by_key[spec.key])
+        lambda spec: spec.replace_attributes(metadata=metadata_by_key[spec.key])
     )
 
 
@@ -279,7 +279,7 @@ def _convert_local_path_to_git_path_single_definition(
         }
 
     return assets_def.map_asset_specs(
-        lambda spec: spec._replace(metadata=metadata_by_key[spec.key])
+        lambda spec: spec.replace_attributes(metadata=metadata_by_key[spec.key])
     )
 
 

python_modules/libraries/dagster-sigma/dagster_sigma_tests/test_translator.py

@@ -80,7 +80,7 @@ def get_asset_key(self, data: SigmaDataset) -> AssetKey:
         def get_asset_spec(self, data: SigmaDataset) -> AssetSpec:
             spec = super().get_asset_spec(data)
             if isinstance(data, SigmaDataset):
-                return spec._replace(description="Custom description")
+                return spec.replace_attributes(description="Custom description")
             return spec
 
     sample_dataset = SigmaDataset(