diff --git a/python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py b/python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py index fcce1b3492fd5..cba83faadd225 100644 --- a/python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py +++ b/python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py @@ -141,18 +141,6 @@ def compute_intersection_with_partition_keys( ) -> "AssetSlice": """Return a new AssetSlice with only the given partition keys if they are in the slice.""" return self._asset_graph_view.compute_intersection_with_partition_keys(partition_keys, self) - partitions_def = check.not_none(self._partitions_def, "Must have partitions def") - for partition_key in partition_keys: - if not partitions_def.has_partition_key(partition_key): - check.failed( - f"Partition key {partition_key} not in partitions def {self._partitions_def}" - ) - - return _slice_from_subset( - self._asset_graph_view, - self._compatible_subset - & AssetSubset.from_partition_keys(self.asset_key, partitions_def, partition_keys), - ) class AssetGraphView: @@ -291,7 +279,11 @@ def compute_intersection_with_partition_keys( self._get_partitions_def(asset_slice.asset_key), "Must have partitions def" ) for partition_key in partition_keys: - if not partitions_def.has_partition_key(partition_key): + if not partitions_def.has_partition_key( + partition_key, + current_time=self.effective_dt, + dynamic_partitions_store=self._queryer, + ): check.failed( f"Partition key {partition_key} not in partitions def {partitions_def}" )