From 5ddc747557bf23c31ed715870a305835a32698b9 Mon Sep 17 00:00:00 2001 From: Christopher DeCarolis Date: Thu, 19 Dec 2024 08:15:35 -0800 Subject: [PATCH] [module-loaders] Make asset check coercion explicit in with_attributes (#26529) ## Summary & Motivation This was one of the most confusing parts of the original code path; that asset check key changing was happening _implicitly_ within with_attributes when you changed output_asset_keys. Instead, provide a top-level, explicit way to remap asset check keys. I think the old behavior was highly mysterious. I found myself questioning what the heck was happening. This is much more; "what it says on the tin." ## How I Tested These Changes Existing tests. --- .../_core/definitions/asset_check_spec.py | 3 + .../dagster/_core/definitions/asset_graph.py | 6 +- .../dagster/_core/definitions/asset_key.py | 6 ++ .../dagster/_core/definitions/assets.py | 32 ++++----- .../_core/definitions/cacheable_assets.py | 67 ++++--------------- .../definitions/load_assets_from_modules.py | 29 +++++--- .../asset_defs_tests/test_asset_job.py | 6 +- .../asset_defs_tests/test_assets.py | 34 ++++------ .../test_unresolved_asset_job.py | 8 +-- .../test_cacheable_assets_defs.py | 26 ++++--- 10 files changed, 94 insertions(+), 123 deletions(-) diff --git a/python_modules/dagster/dagster/_core/definitions/asset_check_spec.py b/python_modules/dagster/dagster/_core/definitions/asset_check_spec.py index b25243bda00f6..99ca04779d492 100644 --- a/python_modules/dagster/dagster/_core/definitions/asset_check_spec.py +++ b/python_modules/dagster/dagster/_core/definitions/asset_check_spec.py @@ -116,3 +116,6 @@ def get_python_identifier(self) -> str: @property def key(self) -> AssetCheckKey: return AssetCheckKey(self.asset_key, self.name) + + def replace_key(self, key: AssetCheckKey) -> "AssetCheckSpec": + return self._replace(asset_key=key.asset_key, name=key.name) diff --git a/python_modules/dagster/dagster/_core/definitions/asset_graph.py b/python_modules/dagster/dagster/_core/definitions/asset_graph.py index fdf5b528385eb..27d88e41fd43b 100644 --- a/python_modules/dagster/dagster/_core/definitions/asset_graph.py +++ b/python_modules/dagster/dagster/_core/definitions/asset_graph.py @@ -204,7 +204,7 @@ def normalize_assets( # AssetKey not subject to any further manipulation. resolved_deps = ResolvedAssetDependencies(assets_defs, []) - input_asset_key_replacements = [ + asset_key_replacements = [ { raw_key: normalized_key for input_name, raw_key in ad.keys_by_input_name.items() @@ -218,8 +218,8 @@ def normalize_assets( # Only update the assets defs if we're actually replacing input asset keys assets_defs = [ - ad.with_attributes(input_asset_key_replacements=reps) if reps else ad - for ad, reps in zip(assets_defs, input_asset_key_replacements) + ad.with_attributes(asset_key_replacements=reps) if reps else ad + for ad, reps in zip(assets_defs, asset_key_replacements) ] # Create unexecutable external assets definitions for any referenced keys for which no diff --git a/python_modules/dagster/dagster/_core/definitions/asset_key.py b/python_modules/dagster/dagster/_core/definitions/asset_key.py index 70f691cdd60c3..607f2f420432b 100644 --- a/python_modules/dagster/dagster/_core/definitions/asset_key.py +++ b/python_modules/dagster/dagster/_core/definitions/asset_key.py @@ -256,6 +256,12 @@ def from_db_string(db_string: str) -> Optional["AssetCheckKey"]: def to_db_string(self) -> str: return seven.json.dumps({"asset_key": self.asset_key.to_string(), "check_name": self.name}) + def with_asset_key_prefix(self, prefix: CoercibleToAssetKeyPrefix) -> "AssetCheckKey": + return AssetCheckKey(self.asset_key.with_prefix(prefix), self.name) + + def replace_asset_key(self, asset_key: AssetKey) -> "AssetCheckKey": + return AssetCheckKey(asset_key, self.name) + EntityKey = Union[AssetKey, AssetCheckKey] T_EntityKey = TypeVar("T_EntityKey", AssetKey, AssetCheckKey, EntityKey) diff --git a/python_modules/dagster/dagster/_core/definitions/assets.py b/python_modules/dagster/dagster/_core/definitions/assets.py index e273bdaa86c65..47f16906d1a8b 100644 --- a/python_modules/dagster/dagster/_core/definitions/assets.py +++ b/python_modules/dagster/dagster/_core/definitions/assets.py @@ -1208,8 +1208,7 @@ def coerce_to_checks_def(self) -> "AssetChecksDefinition": def with_attributes( self, *, - output_asset_key_replacements: Mapping[AssetKey, AssetKey] = {}, - input_asset_key_replacements: Mapping[AssetKey, AssetKey] = {}, + asset_key_replacements: Mapping[AssetKey, AssetKey] = {}, group_names_by_key: Mapping[AssetKey, str] = {}, tags_by_key: Mapping[AssetKey, Mapping[str, str]] = {}, freshness_policy: Optional[ @@ -1254,16 +1253,13 @@ def update_replace_dict_and_conflicts( default_value=DEFAULT_GROUP_NAME, ) - if key in output_asset_key_replacements: - replace_dict["key"] = output_asset_key_replacements[key] + if key in asset_key_replacements: + replace_dict["key"] = asset_key_replacements[key] - if input_asset_key_replacements or output_asset_key_replacements: + if asset_key_replacements: new_deps = [] for dep in spec.deps: - replacement_key = input_asset_key_replacements.get( - dep.asset_key, - output_asset_key_replacements.get(dep.asset_key), - ) + replacement_key = asset_key_replacements.get(dep.asset_key, dep.asset_key) if replacement_key is not None: new_deps.append(dep._replace(asset_key=replacement_key)) else: @@ -1280,33 +1276,31 @@ def update_replace_dict_and_conflicts( ) check_specs_by_output_name = { - output_name: check_spec._replace( - asset_key=output_asset_key_replacements.get( - check_spec.asset_key, check_spec.asset_key + output_name: check_spec.replace_key( + key=check_spec.key.replace_asset_key( + asset_key_replacements.get(check_spec.asset_key, check_spec.asset_key) ) ) for output_name, check_spec in self.node_check_specs_by_output_name.items() } selected_asset_check_keys = { - check_key._replace( - asset_key=output_asset_key_replacements.get( - check_key.asset_key, check_key.asset_key - ) + check_key.replace_asset_key( + asset_key_replacements.get(check_key.asset_key, check_key.asset_key) ) for check_key in self.check_keys } replaced_attributes = dict( keys_by_input_name={ - input_name: input_asset_key_replacements.get(key, key) + input_name: asset_key_replacements.get(key, key) for input_name, key in self.node_keys_by_input_name.items() }, keys_by_output_name={ - output_name: output_asset_key_replacements.get(key, key) + output_name: asset_key_replacements.get(key, key) for output_name, key in self.node_keys_by_output_name.items() }, - selected_asset_keys={output_asset_key_replacements.get(key, key) for key in self.keys}, + selected_asset_keys={asset_key_replacements.get(key, key) for key in self.keys}, backfill_policy=backfill_policy if backfill_policy else self.backfill_policy, is_subset=self.is_subset, check_specs_by_output_name=check_specs_by_output_name, diff --git a/python_modules/dagster/dagster/_core/definitions/cacheable_assets.py b/python_modules/dagster/dagster/_core/definitions/cacheable_assets.py index 5445636a14ed6..70ea705998d90 100644 --- a/python_modules/dagster/dagster/_core/definitions/cacheable_assets.py +++ b/python_modules/dagster/dagster/_core/definitions/cacheable_assets.py @@ -153,8 +153,7 @@ def with_resources( def with_attributes( self, - output_asset_key_replacements: Optional[Mapping[AssetKey, AssetKey]] = None, - input_asset_key_replacements: Optional[Mapping[AssetKey, AssetKey]] = None, + asset_key_replacements: Optional[Mapping[AssetKey, AssetKey]] = None, group_names_by_key: Optional[Mapping[AssetKey, str]] = None, freshness_policy: Optional[ Union[FreshnessPolicy, Mapping[AssetKey, FreshnessPolicy]] @@ -162,8 +161,7 @@ def with_attributes( ) -> "CacheableAssetsDefinition": return PrefixOrGroupWrappedCacheableAssetsDefinition( self, - output_asset_key_replacements=output_asset_key_replacements, - input_asset_key_replacements=input_asset_key_replacements, + asset_key_replacements=asset_key_replacements, group_names_by_key=group_names_by_key, freshness_policy=freshness_policy, ) @@ -247,8 +245,7 @@ class PrefixOrGroupWrappedCacheableAssetsDefinition(WrappedCacheableAssetsDefini def __init__( self, wrapped: CacheableAssetsDefinition, - output_asset_key_replacements: Optional[Mapping[AssetKey, AssetKey]] = None, - input_asset_key_replacements: Optional[Mapping[AssetKey, AssetKey]] = None, + asset_key_replacements: Optional[Mapping[AssetKey, AssetKey]] = None, group_names_by_key: Optional[Mapping[AssetKey, str]] = None, group_name_for_all_assets: Optional[str] = None, prefix_for_all_assets: Optional[List[str]] = None, @@ -260,8 +257,7 @@ def __init__( ] = None, backfill_policy: Optional[BackfillPolicy] = None, ): - self._output_asset_key_replacements = output_asset_key_replacements or {} - self._input_asset_key_replacements = input_asset_key_replacements or {} + self._asset_key_replacements = asset_key_replacements or {} self._group_names_by_key = group_names_by_key or {} self._group_name_for_all_assets = group_name_for_all_assets self._prefix_for_all_assets = prefix_for_all_assets @@ -274,12 +270,8 @@ def __init__( "Cannot set both group_name_for_all_assets and group_names_by_key", ) check.invariant( - not ( - prefix_for_all_assets - and (output_asset_key_replacements or input_asset_key_replacements) - ), - "Cannot set both prefix_for_all_assets and output_asset_key_replacements or" - " input_asset_key_replacements", + not (prefix_for_all_assets and (asset_key_replacements)), + "Cannot set both prefix_for_all_assets and asset_key_replacements", ) super().__init__( @@ -290,22 +282,10 @@ def __init__( def _get_hash(self) -> str: """Generate a stable hash of the various prefix/group mappings.""" contents = hashlib.sha1() - if self._output_asset_key_replacements: + if self._asset_key_replacements: contents.update( _map_to_hashable( - { - tuple(k.path): tuple(v.path) - for k, v in self._output_asset_key_replacements.items() - } - ) - ) - if self._input_asset_key_replacements: - contents.update( - _map_to_hashable( - { - tuple(k.path): tuple(v.path) - for k, v in self._input_asset_key_replacements.items() - } + {tuple(k.path): tuple(v.path) for k, v in self._asset_key_replacements.items()} ) ) if self._group_names_by_key: @@ -330,33 +310,13 @@ def transformed_assets_def(self, assets_def: AssetsDefinition) -> AssetsDefiniti if self._group_name_for_all_assets else self._group_names_by_key ) - output_asset_key_replacements = ( - { - k: AssetKey( - path=( - self._prefix_for_all_assets + list(k.path) - if self._prefix_for_all_assets - else k.path - ) - ) - for k in assets_def.keys - } - if self._prefix_for_all_assets - else self._output_asset_key_replacements - ) - input_asset_key_replacements = ( + asset_key_replacements = ( { - k: AssetKey( - path=( - self._prefix_for_all_assets + list(k.path) - if self._prefix_for_all_assets - else k.path - ) - ) - for k in assets_def.dependency_keys + k: k.with_prefix(self._prefix_for_all_assets) if self._prefix_for_all_assets else k + for k in assets_def.keys | set(assets_def.dependency_keys) } if self._prefix_for_all_assets - else self._input_asset_key_replacements + else self._asset_key_replacements ) if isinstance(self._auto_materialize_policy, dict): automation_condition = { @@ -367,8 +327,7 @@ def transformed_assets_def(self, assets_def: AssetsDefinition) -> AssetsDefiniti else: automation_condition = None return assets_def.with_attributes( - output_asset_key_replacements=output_asset_key_replacements, - input_asset_key_replacements=input_asset_key_replacements, + asset_key_replacements=asset_key_replacements, group_names_by_key=group_names_by_key, freshness_policy=self._freshness_policy, automation_condition=automation_condition, diff --git a/python_modules/dagster/dagster/_core/definitions/load_assets_from_modules.py b/python_modules/dagster/dagster/_core/definitions/load_assets_from_modules.py index bdec14601d014..6b25110097cac 100644 --- a/python_modules/dagster/dagster/_core/definitions/load_assets_from_modules.py +++ b/python_modules/dagster/dagster/_core/definitions/load_assets_from_modules.py @@ -21,6 +21,7 @@ import dagster._check as check from dagster._core.definitions.asset_checks import AssetChecksDefinition, has_only_asset_checks from dagster._core.definitions.asset_key import ( + AssetCheckKey, AssetKey, CoercibleToAssetKeyPrefix, check_opt_coercible_to_asset_key_prefix_param, @@ -429,6 +430,7 @@ def find_modules_in_package(package_module: ModuleType) -> Iterable[ModuleType]: def replace_keys_in_asset( asset: Union[AssetsDefinition, AssetSpec, SourceAsset], key_replacements: Mapping[AssetKey, AssetKey], + check_key_replacements: Mapping[AssetCheckKey, AssetCheckKey], ) -> Union[AssetsDefinition, AssetSpec, SourceAsset]: if isinstance(asset, SourceAsset): return asset.with_attributes(key=key_replacements.get(asset.key, asset.key)) @@ -438,13 +440,7 @@ def replace_keys_in_asset( ) else: updated_object = asset.with_attributes( - output_asset_key_replacements={ - key: key_replacements.get(key, key) - for key in key_iterator(asset, included_targeted_keys=True) - }, - input_asset_key_replacements={ - key: key_replacements.get(key, key) for key in asset.keys_by_input_name.values() - }, + asset_key_replacements=key_replacements, ) return ( updated_object.coerce_to_checks_def() @@ -469,6 +465,10 @@ def assets_defs_and_specs(self) -> Sequence[Union[AssetsDefinition, AssetSpec]]: or isinstance(dagster_object, AssetSpec) ] + @cached_property + def assets_defs(self) -> Sequence[AssetsDefinition]: + return [asset for asset in self.loaded_objects if isinstance(asset, AssetsDefinition)] + @cached_property def checks_defs(self) -> Sequence[AssetChecksDefinition]: return [ @@ -511,12 +511,21 @@ def assets_with_loadable_prefix( for asset_object in self.assets_defs_specs_and_checks_defs for key in key_iterator(asset_object, included_targeted_keys=True) } + all_check_keys = { + check_key for asset_object in self.assets_defs for check_key in asset_object.check_keys + } + key_replacements = {key: key.with_prefix(key_prefix) for key in all_asset_keys} + check_key_replacements = { + check_key: check_key.with_asset_key_prefix(key_prefix) for check_key in all_check_keys + } for asset_object in self.loaded_objects: if isinstance(asset_object, CacheableAssetsDefinition): result_list.append(asset_object.with_prefix_for_all(key_prefix)) elif isinstance(asset_object, AssetsDefinition): - result_list.append(replace_keys_in_asset(asset_object, key_replacements)) + result_list.append( + replace_keys_in_asset(asset_object, key_replacements, check_key_replacements) + ) else: # We don't replace the key for SourceAssets. result_list.append(asset_object) @@ -532,7 +541,9 @@ def assets_with_source_prefix( } for asset_object in self.loaded_objects: if isinstance(asset_object, KeyScopedAssetObjects): - result_list.append(replace_keys_in_asset(asset_object, key_replacements)) + result_list.append( + replace_keys_in_asset(asset_object, key_replacements, check_key_replacements={}) + ) else: result_list.append(asset_object) return ResolvedAssetObjectList(result_list) diff --git a/python_modules/dagster/dagster_tests/asset_defs_tests/test_asset_job.py b/python_modules/dagster/dagster_tests/asset_defs_tests/test_asset_job.py index 0e96216bd1477..99d8733f7e7f4 100644 --- a/python_modules/dagster/dagster_tests/asset_defs_tests/test_asset_job.py +++ b/python_modules/dagster/dagster_tests/asset_defs_tests/test_asset_job.py @@ -2363,10 +2363,10 @@ def test_asset_group_build_subset_job(job_selection, expected_assets, use_multi, for prefix in reversed(prefixes or []): all_assets = [ assets_def.with_attributes( - input_asset_key_replacements={ - k: k.with_prefix(prefix) for k in assets_def.keys_by_input_name.values() + asset_key_replacements={ + k: k.with_prefix(prefix) + for k in set(assets_def.keys_by_input_name.values()) | set(assets_def.keys) }, - output_asset_key_replacements={k: k.with_prefix(prefix) for k in assets_def.keys}, ) for assets_def in all_assets ] diff --git a/python_modules/dagster/dagster_tests/asset_defs_tests/test_assets.py b/python_modules/dagster/dagster_tests/asset_defs_tests/test_assets.py index 5fbc36e955cbb..db853ff7499ea 100644 --- a/python_modules/dagster/dagster_tests/asset_defs_tests/test_assets.py +++ b/python_modules/dagster/dagster_tests/asset_defs_tests/test_assets.py @@ -66,11 +66,9 @@ def asset1(input1, input2): assert input2 replaced = asset1.with_attributes( - output_asset_key_replacements={ - AssetKey(["asset1"]): AssetKey(["prefix1", "asset1_changed"]) - }, - input_asset_key_replacements={ - AssetKey(["something_else", "input2"]): AssetKey(["apple", "banana"]) + asset_key_replacements={ + AssetKey(["asset1"]): AssetKey(["prefix1", "asset1_changed"]), + AssetKey(["something_else", "input2"]): AssetKey(["apple", "banana"]), }, ) @@ -166,9 +164,7 @@ def test_retain_group(): def bar(): pass - replaced = bar.with_attributes( - output_asset_key_replacements={AssetKey(["bar"]): AssetKey(["baz"])} - ) + replaced = bar.with_attributes(asset_key_replacements={AssetKey(["bar"]): AssetKey(["baz"])}) assert replaced.specs_by_key[AssetKey("baz")].group_name == "foo" @@ -179,9 +175,7 @@ def test_retain_freshness_policy(): def bar(): pass - replaced = bar.with_attributes( - output_asset_key_replacements={AssetKey(["bar"]): AssetKey(["baz"])} - ) + replaced = bar.with_attributes(asset_key_replacements={AssetKey(["bar"]): AssetKey(["baz"])}) assert ( replaced.specs_by_key[AssetKey(["baz"])].freshness_policy == bar.specs_by_key[AssetKey(["bar"])].freshness_policy @@ -216,7 +210,7 @@ def my_graph(): ) replaced = my_graph_asset.with_attributes( - output_asset_key_replacements={ + asset_key_replacements={ AssetKey("a"): AssetKey("aa"), AssetKey("b"): AssetKey("bb"), AssetKey("c"): AssetKey("cc"), @@ -245,7 +239,7 @@ def bar(): original = AssetsDefinition.from_graph(bar, metadata_by_output_name={"result": md}) replaced = original.with_attributes( - output_asset_key_replacements={AssetKey(["bar"]): AssetKey(["baz"])} + asset_key_replacements={AssetKey(["bar"]): AssetKey(["baz"])} ) assert ( replaced.specs_by_key[AssetKey(["baz"])].metadata @@ -281,7 +275,7 @@ def bar_(input_last): assert isinstance(bar_.get_partition_mapping(AssetKey(["input_last"])), LastPartitionMapping) replaced = bar_.with_attributes( - input_asset_key_replacements={ + asset_key_replacements={ AssetKey(["input_last"]): AssetKey(["input_last2"]), } ) @@ -305,8 +299,10 @@ def abc_(context, in1, in2, in3): pass replaced_1 = abc_.with_attributes( - output_asset_key_replacements={AssetKey(["a"]): AssetKey(["foo", "foo_a"])}, - input_asset_key_replacements={AssetKey(["in1"]): AssetKey(["foo", "bar_in1"])}, + asset_key_replacements={ + AssetKey(["a"]): AssetKey(["foo", "foo_a"]), + AssetKey(["in1"]): AssetKey(["foo", "bar_in1"]), + }, ) assert replaced_1.keys == {AssetKey(["foo", "foo_a"]), AssetKey("b"), AssetKey("c")} @@ -328,11 +324,9 @@ def abc_(context, in1, in2, in3): assert subbed_1.keys == {AssetKey(["foo", "foo_a"]), AssetKey("b")} replaced_2 = subbed_1.with_attributes( - output_asset_key_replacements={ + asset_key_replacements={ AssetKey(["foo", "foo_a"]): AssetKey(["again", "foo", "foo_a"]), AssetKey(["b"]): AssetKey(["something", "bar_b"]), - }, - input_asset_key_replacements={ AssetKey(["foo", "bar_in1"]): AssetKey(["again", "foo", "bar_in1"]), AssetKey(["in2"]): AssetKey(["foo", "in2"]), AssetKey(["in3"]): AssetKey(["foo", "in3"]), @@ -2220,7 +2214,7 @@ def my_multi_asset(): assert my_multi_asset.specs_by_key[AssetKey("out2")].owners == ["user@dagsterlabs.com"] prefixed_asset = my_multi_asset.with_attributes( - output_asset_key_replacements={AssetKey(["out1"]): AssetKey(["prefix", "out1"])} + asset_key_replacements={AssetKey(["out1"]): AssetKey(["prefix", "out1"])} ) assert prefixed_asset.specs_by_key[AssetKey(["prefix", "out1"])].owners == [ "user@dagsterlabs.com" diff --git a/python_modules/dagster/dagster_tests/asset_defs_tests/test_unresolved_asset_job.py b/python_modules/dagster/dagster_tests/asset_defs_tests/test_unresolved_asset_job.py index 51f4917ae23b5..c8d475662a658 100644 --- a/python_modules/dagster/dagster_tests/asset_defs_tests/test_unresolved_asset_job.py +++ b/python_modules/dagster/dagster_tests/asset_defs_tests/test_unresolved_asset_job.py @@ -355,11 +355,9 @@ def test_define_selection_job(job_selection, expected_assets, use_multi, prefixe for prefix in reversed(prefixes or []): prefixed_assets = [ assets_def.with_attributes( - input_asset_key_replacements={ - key: key.with_prefix(prefix) for key in assets_def.keys_by_input_name.values() - }, - output_asset_key_replacements={ - key: key.with_prefix(prefix) for key in assets_def.keys + asset_key_replacements={ + key: key.with_prefix(prefix) + for key in set(assets_def.keys_by_input_name.values()) | set(assets_def.keys) }, ) for assets_def in prefixed_assets diff --git a/python_modules/dagster/dagster_tests/general_tests/test_cacheable_assets_defs.py b/python_modules/dagster/dagster_tests/general_tests/test_cacheable_assets_defs.py index 1ec97dc41eca1..6391f6e83cbc7 100644 --- a/python_modules/dagster/dagster_tests/general_tests/test_cacheable_assets_defs.py +++ b/python_modules/dagster/dagster_tests/general_tests/test_cacheable_assets_defs.py @@ -1,4 +1,4 @@ -from typing import Sequence +from typing import List, Sequence, Union, cast import dagster._check as check import pytest @@ -151,7 +151,9 @@ def test_resolve_wrong_data(): recon_repo.get_definition() -def define_uncacheable_and_resource_dependent_cacheable_assets(): +def define_uncacheable_and_resource_dependent_cacheable_assets() -> ( + Sequence[Union[CacheableAssetsDefinition, AssetsDefinition]] +): class ResourceDependentCacheableAsset(CacheableAssetsDefinition): def __init__(self): super().__init__("res_midstream") @@ -295,7 +297,7 @@ def _op(): ) -def test_multiple_wrapped_cached_assets(): +def test_multiple_wrapped_cached_assets() -> None: """Test that multiple wrappers (with_attributes, with_resources) work properly on cacheable assets.""" @resource @@ -304,9 +306,7 @@ def foo_resource(): my_cacheable_assets_with_group_and_asset = [ x.with_attributes( - output_asset_key_replacements={ - AssetKey("res_downstream"): AssetKey("res_downstream_too") - } + asset_key_replacements={AssetKey("res_downstream"): AssetKey("res_downstream_too")} ) for x in with_resources( [ @@ -333,13 +333,19 @@ def resource_dependent_repo_with_resources(): assert isinstance(repo.get_job("all_asset_job"), JobDefinition) my_cool_group_sel = AssetSelection.groups("my_cool_group") + cacheable_resource_asset = cast( + CacheableAssetsDefinition, my_cacheable_assets_with_group_and_asset[0] + ) + resolved_defs = list( + cacheable_resource_asset.build_definitions( + cacheable_resource_asset.compute_cacheable_data() + ) + ) assert ( len( my_cool_group_sel.resolve( - my_cacheable_assets_with_group_and_asset[0].build_definitions( - my_cacheable_assets_with_group_and_asset[0].compute_cacheable_data() - ) - + my_cacheable_assets_with_group_and_asset[1:] + resolved_defs + + cast(List[AssetsDefinition], my_cacheable_assets_with_group_and_asset[1:]) ) ) == 1