File size: 3,296 Bytes
3e93464 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | from copy import deepcopy
from app.models.workflow import (
OptimizationResponse,
OptimizationSuggestion,
WorkflowDocument,
)
class WorkflowOptimizer:
def optimize(self, workflow: WorkflowDocument) -> OptimizationResponse:
optimized = deepcopy(workflow)
suggestions: list[OptimizationSuggestion] = []
http_nodes = [
node for node in optimized.nodes if node.data.type == "n8n-nodes-base.httpRequest"
]
for node in http_nodes:
options = node.data.parameters.setdefault("options", {})
if isinstance(options, dict) and "retry" not in options:
options.update({"retry": {"maxTries": 3, "waitBetweenTries": 1000}})
suggestions.append(
OptimizationSuggestion(
title="Add API retries",
description=f"Added bounded retry behavior to {node.data.label}.",
impact="high",
nodeIds=[node.id],
)
)
if not optimized.settings.get("saveExecutionProgress"):
optimized.settings["saveExecutionProgress"] = True
suggestions.append(
OptimizationSuggestion(
title="Enable execution recovery",
description="Save execution progress so long-running workflows can recover.",
impact="medium",
)
)
linear_pairs = []
for edge in optimized.edges:
source = next((node for node in optimized.nodes if node.id == edge.source), None)
target = next((node for node in optimized.nodes if node.id == edge.target), None)
if source and target and source.data.type == target.data.type == "n8n-nodes-base.set":
linear_pairs.append((source, target))
if linear_pairs:
suggestions.append(
OptimizationSuggestion(
title="Combine adjacent Edit Fields nodes",
description="Adjacent field transformations can be handled by one node.",
impact="medium",
nodeIds=[node.id for pair in linear_pairs for node in pair],
)
)
branches = {}
for edge in optimized.edges:
branches[edge.source] = branches.get(edge.source, 0) + 1
parallel = [node_id for node_id, count in branches.items() if count > 1]
if parallel:
suggestions.append(
OptimizationSuggestion(
title="Review parallel branches",
description="Independent branches can execute concurrently; merge only when required.",
impact="medium",
nodeIds=parallel,
)
)
if not suggestions:
suggestions.append(
OptimizationSuggestion(
title="Workflow is already compact",
description="No deterministic optimization could be applied safely.",
impact="low",
)
)
return OptimizationResponse(workflow=optimized, suggestions=suggestions)
optimizer = WorkflowOptimizer()
|