Remove @cached_method

dagster-io · Oct 22, 2024 · 05c2e02 · 05c2e02
1 parent 620e7de
commit 05c2e02
Showing 1 changed file with 2 additions and 10 deletions.
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
@@ -32,7 +32,6 @@
 from dagster._core.loader import LoadingContext
 from dagster._time import get_current_datetime
 from dagster._utils.aiodataloader import DataLoader
-from dagster._utils.cached_method import cached_method
 
 if TYPE_CHECKING:
     from dagster._core.definitions.base_asset_graph import BaseAssetGraph, BaseAssetNode
@@ -171,7 +170,6 @@ def _get_partitions_def(self, key: T_EntityKey) -> Optional["PartitionsDefinitio
         else:
             return None
 
-    @cached_method
     def get_full_subset(self, *, key: T_EntityKey) -> EntitySubset[T_EntityKey]:
         partitions_def = self._get_partitions_def(key)
         value = (
@@ -185,7 +183,6 @@ def get_full_subset(self, *, key: T_EntityKey) -> EntitySubset[T_EntityKey]:
         )
         return EntitySubset(self, key=key, value=_ValidatedEntitySubsetValue(value))
 
-    @cached_method
     def get_empty_subset(self, *, key: T_EntityKey) -> EntitySubset[T_EntityKey]:
         partitions_def = self._get_partitions_def(key)
         value = partitions_def.empty_subset() if partitions_def else False
@@ -363,7 +360,7 @@ async def compute_subset_with_status(
         """Returns the subset of an asset check that matches a given status."""
         from dagster._core.storage.event_log.base import AssetCheckSummaryRecord
 
-        summary = AssetCheckSummaryRecord.blocking_get(self, key)
+        summary = await AssetCheckSummaryRecord.gen(self, key)
         latest_record = summary.last_check_execution_record if summary else None
         resolved_status = (
             latest_record.resolve_status(self)
@@ -479,15 +476,13 @@ async def _compute_missing_asset_subset(
                 key=key, asset_partitions=missing_asset_partitions
             )
 
-    @cached_method
     async def compute_run_in_progress_subset(self, *, key: EntityKey) -> EntitySubset:
         return await _dispatch(
             key=key,
             check_method=self._compute_run_in_progress_check_subset,
             asset_method=self._compute_run_in_progress_asset_subset,
         )
 
-    @cached_method
     async def compute_backfill_in_progress_subset(self, *, key: EntityKey) -> EntitySubset:
         async def get_empty_subset(key: EntityKey) -> EntitySubset:
             return self.get_empty_subset(key=key)
@@ -499,15 +494,13 @@ async def get_empty_subset(key: EntityKey) -> EntitySubset:
             asset_method=self._compute_backfill_in_progress_asset_subset,
         )
 
-    @cached_method
     async def compute_execution_failed_subset(self, *, key: EntityKey) -> EntitySubset:
         return await _dispatch(
             key=key,
             check_method=self._compute_execution_failed_check_subset,
             asset_method=self._compute_execution_failed_asset_subset,
         )
 
-    @cached_method
     async def compute_missing_subset(
         self, *, key: EntityKey, from_subset: EntitySubset
     ) -> EntitySubset:
@@ -614,7 +607,7 @@ async def _compute_updated_since_time_subset(
         from dagster._core.storage.event_log.base import AssetCheckSummaryRecord
 
         # intentionally left unimplemented for AssetKey, as this is a less performant query
-        summary = AssetCheckSummaryRecord.blocking_get(self, key)
+        summary = await AssetCheckSummaryRecord.gen(self, key)
         record = summary.last_check_execution_record if summary else None
         if (
             record is None
@@ -626,7 +619,6 @@ async def _compute_updated_since_time_subset(
         else:
             return self.get_full_subset(key=key)
 
-    @cached_method
     async def compute_updated_since_temporal_context_subset(
         self, *, key: EntityKey, temporal_context: TemporalContext
     ) -> EntitySubset: