From 4f7d59666a962ea123fd218c580006dec10fb1f5 Mon Sep 17 00:00:00 2001 From: Dariusz L Date: Sat, 8 Aug 2026 19:05:14 +0200 Subject: [PATCH] refactor: share workflow cache helpers --- core/workflow/analysis.py | 41 ++++++++++++++++++++----------- core/workflow/inventory.py | 18 ++++---------- tests/test_workflow_components.py | 24 ++++++++++++++++++ 3 files changed, 56 insertions(+), 27 deletions(-) diff --git a/core/workflow/analysis.py b/core/workflow/analysis.py index c51d04d..f833e2d 100644 --- a/core/workflow/analysis.py +++ b/core/workflow/analysis.py @@ -37,24 +37,41 @@ def _normalize_workflow_analysis_value(value: Any) -> Any: return value +def _hash_normalized_workflow_value(value: Any) -> str: + serialized_value = json.dumps( + _normalize_workflow_analysis_value(value), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(serialized_value.encode("utf-8")).hexdigest() + + +def _get_workflow_scope_key( + node_id: Any, + *, + subgraph_id: Any = "", + is_top_level: bool, +) -> tuple[str, str, str]: + scope = "top" if is_top_level else "subgraph" + return (scope, "" if is_top_level else str(subgraph_id or ""), str(node_id)) + + def _get_workflow_node_cache_key( node: Dict[str, Any], *, subgraph_id: Any = "", is_top_level: bool, ) -> tuple[str, str, str]: - scope = "top" if is_top_level else "subgraph" - return (scope, "" if is_top_level else str(subgraph_id or ""), str(node.get("id"))) + return _get_workflow_scope_key( + node.get("id"), + subgraph_id=subgraph_id, + is_top_level=is_top_level, + ) def _get_workflow_node_fingerprint(node: Dict[str, Any]) -> str: - serialized_node = json.dumps( - _normalize_workflow_analysis_value(node), - ensure_ascii=False, - sort_keys=True, - separators=(",", ":"), - ) - return hashlib.sha256(serialized_node.encode("utf-8")).hexdigest() + return _hash_normalized_workflow_value(node) def _get_workflow_node_fingerprints( @@ -176,16 +193,12 @@ def _get_promoted_widget_context_cache_key( _normalize_workflow_analysis_value(node) ) - serialized_context = json.dumps( + return _hash_normalized_workflow_value( { "subgraphs": context_subgraphs, "instances": context_instances, }, - ensure_ascii=False, - sort_keys=True, - separators=(",", ":"), ) - return hashlib.sha256(serialized_context.encode("utf-8")).hexdigest() def analyze_workflow_models( diff --git a/core/workflow/inventory.py b/core/workflow/inventory.py index 0f7c8e1..49a27fc 100644 --- a/core/workflow/inventory.py +++ b/core/workflow/inventory.py @@ -1,7 +1,5 @@ """Shared workflow model inventory and incremental cache management.""" -import hashlib -import json import threading import time from collections import OrderedDict @@ -28,10 +26,10 @@ def _build_workflow_node_cache( for ref in model_refs: is_top_level = ref.get("is_top_level") is not False - key = ( - "top" if is_top_level else "subgraph", - "" if is_top_level else str(ref.get("subgraph_id") or ""), - str(ref.get("node_id")), + key = analysis._get_workflow_scope_key( + ref.get("node_id"), + subgraph_id=ref.get("subgraph_id"), + is_top_level=is_top_level, ) if key in refs_by_node: refs_by_node[key].append(ref) @@ -48,13 +46,7 @@ def _build_workflow_node_cache( def _get_workflow_model_inventory_cache_key( workflow_json: Dict[str, Any], ) -> str: - serialized_workflow = json.dumps( - analysis._normalize_workflow_analysis_value(workflow_json), - ensure_ascii=False, - sort_keys=True, - separators=(",", ":"), - ) - return hashlib.sha256(serialized_workflow.encode("utf-8")).hexdigest() + return analysis._hash_normalized_workflow_value(workflow_json) def _get_analysis_log_context( diff --git a/tests/test_workflow_components.py b/tests/test_workflow_components.py index 2fb74dd..0b7eb2a 100644 --- a/tests/test_workflow_components.py +++ b/tests/test_workflow_components.py @@ -57,6 +57,30 @@ def test_workflow_node_traversal_preserves_node_order_and_scope(): assert contexts[3].subgraph_id == "group-1" +def test_workflow_cache_helpers_share_scope_keys_and_normalized_hashing(): + node = {"id": 2, "type": "Inner", "pos": [10, 20]} + + assert analysis._get_workflow_node_cache_key( + node, + subgraph_id="group-1", + is_top_level=False, + ) == analysis._get_workflow_scope_key( + 2, + subgraph_id="group-1", + is_top_level=False, + ) + assert analysis._get_workflow_node_cache_key( + node, + is_top_level=True, + ) == ("top", "", "2") + assert analysis._get_workflow_node_fingerprint(node) == ( + analysis._hash_normalized_workflow_value(node) + ) + assert analysis._get_workflow_node_fingerprint( + {"type": "Inner", "id": 2, "pos": [999, 999]} + ) == analysis._get_workflow_node_fingerprint(node) + + def test_active_workflow_node_traversal_follows_nested_referenced_subgraphs_only(): workflow = { "nodes": [