From c63f9ec0cef11612eb9e16cf91f83d5b3d6984a1 Mon Sep 17 00:00:00 2001 From: briantu Date: Fri, 11 Oct 2024 14:59:35 -0400 Subject: [PATCH] Sum over gathered --- .../automation_condition_evaluator.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/python_modules/dagster/dagster/_core/definitions/declarative_automation/automation_condition_evaluator.py b/python_modules/dagster/dagster/_core/definitions/declarative_automation/automation_condition_evaluator.py index 3f30d3f048ef6..2285eb742aaf5 100644 --- a/python_modules/dagster/dagster/_core/definitions/declarative_automation/automation_condition_evaluator.py +++ b/python_modules/dagster/dagster/_core/definitions/declarative_automation/automation_condition_evaluator.py @@ -112,9 +112,7 @@ async def async_evaluate( num_conditions = len(self.entity_keys) num_evaluated = 0 - async def _evaluate_entity_async(entity_key: EntityKey) -> None: - nonlocal num_evaluated - + async def _evaluate_entity_async(entity_key: EntityKey) -> int: self.logger.debug( f"Evaluating {entity_key.to_user_string()} ({num_evaluated+1}/{num_conditions})" ) @@ -138,7 +136,7 @@ async def _evaluate_entity_async(entity_key: EntityKey) -> None: f"requested ({requested_str}) " f"({format(result.end_timestamp - result.start_timestamp, '.3f')} seconds)" ) - num_evaluated += 1 + return 1 for topo_level in self.asset_graph.toposorted_entity_keys_by_level: coroutines = [ @@ -146,7 +144,8 @@ async def _evaluate_entity_async(entity_key: EntityKey) -> None: for entity_key in topo_level if entity_key in self.entity_keys ] - await asyncio.gather(*coroutines) + gathered = await asyncio.gather(*coroutines) + num_evaluated += sum(gathered) return list(self.current_results_by_key.values()), [ v for v in self.request_subsets_by_key.values() if not v.is_empty