From ad87c12c96a31a56ae0f6106f7ce7678e998158d Mon Sep 17 00:00:00 2001
From: gibsondan <daniel@elementl.com>
Date: Fri, 8 Nov 2024 12:10:33 -0600
Subject: [PATCH] Don't iterate the whole asset graph to generate asset
 condition evaluation records (#25809)

BK

## Summary & Motivation

## How I Tested These Changes

## Changelog

> Insert changelog entry or delete this section.
---
 .../fetch_asset_condition_evaluations.py        | 17 +----------------
 .../schema/asset_condition_evaluations.py       |  2 --
 2 files changed, 1 insertion(+), 18 deletions(-)

diff --git a/python_modules/dagster-graphql/dagster_graphql/implementation/fetch_asset_condition_evaluations.py b/python_modules/dagster-graphql/dagster_graphql/implementation/fetch_asset_condition_evaluations.py
index 346d2b56c8771..68231ad9e9fc7 100644
--- a/python_modules/dagster-graphql/dagster_graphql/implementation/fetch_asset_condition_evaluations.py
+++ b/python_modules/dagster-graphql/dagster_graphql/implementation/fetch_asset_condition_evaluations.py
@@ -4,7 +4,6 @@
 from dagster import AssetKey
 from dagster._core.scheduler.instigation import AutoMaterializeAssetEvaluationRecord
 
-from dagster_graphql.implementation.fetch_assets import get_asset_nodes_by_asset_key
 from dagster_graphql.schema.asset_condition_evaluations import (
     GrapheneAssetConditionEvaluation,
     GrapheneAssetConditionEvaluationRecord,
@@ -41,23 +40,9 @@ def _get_graphene_records_from_evaluations(
     graphene_info: "ResolveInfo",
     evaluation_records: Sequence[AutoMaterializeAssetEvaluationRecord],
 ) -> GrapheneAssetConditionEvaluationRecords:
-    asset_keys = {record.key for record in evaluation_records}
-
-    partitions_defs = {}
-
-    nodes = get_asset_nodes_by_asset_key(graphene_info)
-    for asset_key in asset_keys:
-        asset_node = nodes.get(asset_key)
-        partitions_defs[asset_key] = (
-            asset_node.asset_node_snap.partitions.get_partitions_definition()
-            if asset_node and asset_node.asset_node_snap.partitions
-            else None
-        )
-
     return GrapheneAssetConditionEvaluationRecords(
         records=[
-            GrapheneAssetConditionEvaluationRecord(evaluation, partitions_defs[evaluation.key])
-            for evaluation in evaluation_records
+            GrapheneAssetConditionEvaluationRecord(evaluation) for evaluation in evaluation_records
         ]
     )
 
diff --git a/python_modules/dagster-graphql/dagster_graphql/schema/asset_condition_evaluations.py b/python_modules/dagster-graphql/dagster_graphql/schema/asset_condition_evaluations.py
index b12396e5c124b..d5626cc10c1e4 100644
--- a/python_modules/dagster-graphql/dagster_graphql/schema/asset_condition_evaluations.py
+++ b/python_modules/dagster-graphql/dagster_graphql/schema/asset_condition_evaluations.py
@@ -8,7 +8,6 @@
     AutomationConditionEvaluation,
     AutomationConditionSnapshot,
 )
-from dagster._core.definitions.partition import PartitionsDefinition
 from dagster._core.scheduler.instigation import AutoMaterializeAssetEvaluationRecord
 
 from dagster_graphql.implementation.events import iterate_metadata_entries
@@ -272,7 +271,6 @@ class Meta:
     def __init__(
         self,
         record: AutoMaterializeAssetEvaluationRecord,
-        partitions_def: Optional[PartitionsDefinition],
     ):
         evaluation_with_run_ids = record.get_evaluation_with_run_ids()
         root_evaluation = evaluation_with_run_ids.evaluation