Bulk asset editting

examples/experimental/dagster-airlift/examples/dbt-example/dbt_example/dagster_defs/dbt_cloud_assets.py

@@ -1,4 +1,5 @@
-from dagster import AssetsDefinition, multi_asset
+from dagster import AssetsDefinition, AutomationCondition, multi_asset
+from dagster._core.definitions.asset_spec import replace_asset_attributes
 
 from .constants import DBT_UPSTREAMS
 from .dbt_cloud_utils import (
@@ -8,7 +9,6 @@
     get_project,
     relevant_check_specs,
 )
-from .utils import eager
 
 
 def get_dbt_cloud_assets() -> AssetsDefinition:
@@ -17,7 +17,9 @@ def get_dbt_cloud_assets() -> AssetsDefinition:
     filtered_specs = filter_specs_by_tag(dbt_cloud_project.get_asset_specs(), EXPECTED_TAG)
     specs_with_dbt_core_deps = add_deps(DBT_UPSTREAMS, filtered_specs)
     # Dbt cloud assets will run every time the upstream dags run.
-    specs_with_eager_automation = eager(specs_with_dbt_core_deps)
+    specs_with_eager_automation = replace_asset_attributes(
+        specs_with_dbt_core_deps, automation_condition=AutomationCondition.eager()
+    )
 
     @multi_asset(
         specs=specs_with_eager_automation,

examples/experimental/dagster-airlift/examples/dbt-example/dbt_example/dagster_defs/federated_airflow_1.py

@@ -1,6 +1,7 @@
-from typing import Sequence
+from typing import Iterable
 
 from dagster import AssetsDefinition, Definitions, SensorDefinition
+from dagster._core.definitions.asset_spec import replace_asset_attributes
 from dagster_airlift.core import (
     AirflowBasicAuthBackend,
     AirflowInstance,
@@ -13,7 +14,7 @@
     PASSWORD,
     USERNAME,
 )
-from .utils import with_group
+from .utils import all_assets_defs
 
 airflow_instance = AirflowInstance(
     auth_backend=AirflowBasicAuthBackend(
@@ -33,9 +34,6 @@ def get_other_team_airflow_sensor() -> SensorDefinition:
     return next(iter(defs.sensors))
 
 
-def get_other_team_airflow_assets() -> Sequence[AssetsDefinition]:
-    defs = get_federated_airflow_defs()
-    return [
-        with_group(assets_def, "upstream_team_airflow")
-        for assets_def in defs.get_repository_def().assets_defs_by_key.values()
-    ]
+def get_other_team_airflow_assets() -> Iterable[AssetsDefinition]:
+    assets_defs = all_assets_defs(get_federated_airflow_defs())
+    return replace_asset_attributes(assets=assets_defs, group_name="upstream_team_airflow")

examples/experimental/dagster-airlift/examples/dbt-example/dbt_example/dagster_defs/federated_airflow_2.py

@@ -1,14 +1,15 @@
-from typing import Sequence
+from typing import Iterable, Sequence
 
 from dagster import AssetsDefinition, Definitions, SensorDefinition
+from dagster._core.definitions.asset_spec import replace_asset_attributes
 from dagster_airlift.core import (
     AirflowBasicAuthBackend,
     AirflowInstance,
     build_defs_from_airflow_instance,
 )
 
 from .constants import LEGACY_FEDERATED_BASE_URL, LEGACY_FEDERATED_INSTANCE_NAME, PASSWORD, USERNAME
-from .utils import with_group
+from .utils import all_assets_defs
 
 airflow_instance = AirflowInstance(
     auth_backend=AirflowBasicAuthBackend(
@@ -22,15 +23,15 @@ def get_federated_airflow_defs() -> Definitions:
     return build_defs_from_airflow_instance(airflow_instance=airflow_instance)
 
 
+def federated_airflow_assets() -> Sequence[AssetsDefinition]:
+    return all_assets_defs(get_federated_airflow_defs())
+
+
 def get_legacy_instance_airflow_sensor() -> SensorDefinition:
     defs = get_federated_airflow_defs()
     assert defs.sensors
     return next(iter(defs.sensors))
 
 
-def get_legacy_instance_airflow_assets() -> Sequence[AssetsDefinition]:
-    defs = get_federated_airflow_defs()
-    return [
-        with_group(assets_def, "legacy_airflow")
-        for assets_def in defs.get_repository_def().assets_defs_by_key.values()
-    ]
+def get_legacy_instance_airflow_assets() -> Iterable[AssetsDefinition]:
+    return replace_asset_attributes(federated_airflow_assets(), group_name="legacy_airflow")

examples/experimental/dagster-airlift/examples/dbt-example/dbt_example/dagster_defs/jaffle_shop.py

@@ -1,4 +1,5 @@
-from dagster import AssetExecutionContext
+from dagster import AssetExecutionContext, AutomationCondition
+from dagster._core.definitions.asset_spec import replace_asset_attributes
 from dagster_dbt import (
     DagsterDbtTranslator,
     DagsterDbtTranslatorSettings,
@@ -8,7 +9,7 @@
 )
 
 from .constants import DBT_SOURCE_TO_DAG, dbt_manifest_path, dbt_project_path
-from .utils import eager_asset, with_deps
+from .utils import with_deps
 
 
 @dbt_assets(
@@ -23,11 +24,18 @@ def jaffle_shop_assets(context: AssetExecutionContext, dbt: DbtCliResource):
     yield from dbt.cli(["build"], context=context).stream()
 
 
-jaffle_shop_external_assets = [
-    spec._replace(code_version=None, skippable=False) for spec in jaffle_shop_assets.specs
-]
+# jaffle_shop_external_assets = [
+#     spec._replace(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))
+# jaffle_shop_with_upstream = eager_asset(with_deps(DBT_SOURCE_TO_DAG, jaffle_shop_assets))
+
+jaffle_shop_with_upstream = replace_asset_attributes(
+    [with_deps(DBT_SOURCE_TO_DAG, jaffle_shop_assets)],
+    code_version=None,
+    skippable=False,
+    automation_condition=AutomationCondition.eager(),
+)
 
 
 def jaffle_shop_resource() -> DbtCliResource:

examples/experimental/dagster-airlift/examples/dbt-example/dbt_example/dagster_defs/utils.py

@@ -1,46 +1,15 @@
 from typing import Mapping, Sequence
 
-from dagster import AssetsDefinition, AssetSpec, AutomationCondition, Definitions, Nothing
+from dagster import AssetsDefinition, Definitions, Nothing
 from dagster._core.definitions.asset_key import AssetKey
 from dagster._core.definitions.decorators.decorator_assets_definition_builder import (
     stringify_asset_key_to_input_name,
 )
 from dagster._core.definitions.input import In
 
 
-def eager_asset(assets_def: AssetsDefinition) -> AssetsDefinition:
-    return assets_def.map_asset_specs(
-        lambda spec: spec._replace(automation_condition=AutomationCondition.eager())
-        if spec.automation_condition is None
-        else spec
-    )
-
-
-def apply_eager_automation(defs: Definitions) -> Definitions:
-    assets = []
-    for asset in defs.assets or []:
-        if not isinstance(asset, AssetsDefinition):
-            continue
-        if not asset.keys:
-            continue
-        assets.append(
-            asset.map_asset_specs(
-                lambda spec: spec._replace(automation_condition=AutomationCondition.eager())
-                if spec.automation_condition is None
-                else spec
-            )
-        )
-    return Definitions(
-        assets=assets,
-        asset_checks=defs.asset_checks,
-        sensors=defs.sensors,
-        schedules=defs.schedules,
-        resources=defs.resources,
-    )
-
-
-def with_group(assets_def: AssetsDefinition, group_name: str) -> AssetsDefinition:
-    return assets_def.map_asset_specs(lambda spec: spec._replace(group_name=group_name))
+def all_assets_defs(defs: Definitions) -> Sequence[AssetsDefinition]:
+    return list(defs.get_repository_def().assets_defs_by_key.values())
 
 
 def with_deps(
@@ -90,7 +59,3 @@ def with_deps(
         execution_type=assets_def.execution_type,
         auto_materialize_policies_by_key=assets_def.auto_materialize_policies_by_key,
     )
-
-
-def eager(specs: Sequence[AssetSpec]) -> Sequence[AssetSpec]:
-    return [spec._replace(automation_condition=AutomationCondition.eager()) for spec in specs]

python_modules/dagster/dagster/_core/definitions/asset_spec.py

@@ -10,6 +10,7 @@
     Optional,
     Sequence,
     Set,
+    Union,
 )
 
 import dagster._check as check
@@ -20,6 +21,7 @@
     only_allow_hidden_params_in_kwargs,
     public,
 )
+from dagster._core.definitions.assets import AssetsDefinition
 from dagster._core.definitions.auto_materialize_policy import AutoMaterializePolicy
 from dagster._core.definitions.declarative_automation.automation_condition import (
     AutomationCondition,
@@ -294,6 +296,76 @@ def with_io_manager_key(self, io_manager_key: str) -> "AssetSpec":
         )
 
 
+def to_assets_defs(
+    assets: Iterable[Union[AssetSpec, AssetsDefinition]],
+) -> Iterable[AssetsDefinition]:
+    for asset in assets:
+        if isinstance(asset, AssetSpec):
+            yield AssetsDefinition(specs=[asset])
+        else:
+            yield asset
+
+
+def replace_asset_attributes(
+    assets: Iterable[Union[AssetSpec, AssetsDefinition]],
+    *,
+    key: CoercibleToAssetKey = ...,
+    deps: Optional[Iterable["CoercibleToAssetDep"]] = ...,
+    description: Optional[str] = ...,
+    metadata: Optional[Mapping[str, Any]] = ...,
+    skippable: bool = ...,
+    group_name: Optional[str] = ...,
+    code_version: Optional[str] = ...,
+    freshness_policy: Optional[FreshnessPolicy] = ...,
+    automation_condition: Optional[AutomationCondition] = ...,
+    owners: Optional[Sequence[str]] = ...,
+    tags: Optional[Mapping[str, str]] = ...,
+    kinds: Optional[Set[str]] = ...,
+    partitions_def: Optional[PartitionsDefinition] = ...,
+) -> Iterable[AssetsDefinition]:
+    for asset in assets:
+        if isinstance(asset, AssetSpec):
+            yield AssetsDefinition(
+                specs=[
+                    replace_attributes(
+                        asset,
+                        key=key,
+                        deps=deps,
+                        description=description,
+                        metadata=metadata,
+                        skippable=skippable,
+                        group_name=group_name,
+                        code_version=code_version,
+                        freshness_policy=freshness_policy,
+                        automation_condition=automation_condition,
+                        owners=owners,
+                        tags=tags,
+                        kinds=kinds,
+                        partitions_def=partitions_def,
+                    )
+                ]
+            )
+        else:
+            yield asset.map_asset_specs(
+                lambda spec: replace_attributes(
+                    spec,
+                    key=key,
+                    deps=deps,
+                    description=description,
+                    metadata=metadata,
+                    skippable=skippable,
+                    group_name=group_name,
+                    code_version=code_version,
+                    freshness_policy=freshness_policy,
+                    automation_condition=automation_condition,
+                    owners=owners,
+                    tags=tags,
+                    kinds=kinds,
+                    partitions_def=partitions_def,
+                )
+            )
+
+
 def replace_attributes(
     spec: AssetSpec,
     *,