cp

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}"
                 )