⚡️ Speed up function update_edges_with_latest_component_versions by 32% in src/backend/base/langflow/initial_setup/setup.py

[codeflash-ai]

📄 update_edges_with_latest_component_versions in src/backend/base/langflow/initial_setup/setup.py

✨ Performance Summary:

• Speed Increase: 📈 32% (0.32x faster)
• Runtime Reduction: ⏱️ From 802 microseconds down to 609 microseconds (best of 13 runs)

---

📝 Explanation and details

Here's the optimized version of the provided Python program to improve runtime and memory efficiency. The main focus will be on reducing redundant operations and improving the search process for nodes.

Optimizations.

1. Dictionary Lookup.

   • Instead of iterating through the nodes list multiple times, create a dictionary nodes_by_id that maps node IDs to their corresponding node objects. This allows constant-time lookups.

2. Efficient String Replacement.

   • Simplified the string replacement in scape_json_parse.

3. Single-pass Logging.

   • Collect log messages in a more condensed way and join them once, reducing string concatenation overhead.

4. Combined Lookup Logic.

   • Combined the searching logic for output_data into a single line that attempts both possible lookups, reducing complexity.

5. Error Handling.

   • Moved JSON decode error handling to a helper function safe_escape_json_dump to avoid repeating the same try-except logic.

These changes should improve both runtime performance and memory efficiency while maintaining the original functionality.

---

✅ Correctness verification

The new optimized code was tested for correctness. The results are listed below:

Test	Status	Details
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 7 Passed	See below
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Coverage	70.0%

🌀 Generated Regression Tests Details

Click to view details

import json
from collections import defaultdict
from copy import deepcopy
from unittest.mock import patch

# imports
import pytest  # used for our unit tests
from langflow.initial_setup.setup import \
    update_edges_with_latest_component_versions
# Mocking loguru logger
from loguru import logger


# unit tests


def test_no_edges_in_project_data():
    project_data = {
        "nodes": [
            {"id": "1", "data": {"node": {"outputs": [{"name": "out1", "types": ["type1"]}], "display_name": "Node1"}}},
            {"id": "2", "data": {"node": {"template": {"field1": {"input_types": ["type1"]}}, "display_name": "Node2"}}}
        ],
        "edges": []
    }
    codeflash_output = update_edges_with_latest_component_versions(project_data)

def test_no_nodes_in_project_data():
    project_data = {
        "nodes": [],
        "edges": [
            {"source": "1", "target": "2", "data": {"sourceHandle": '{"name": "out1", "output_types": ["old_type"]}', "targetHandle": '{"fieldName": "field1", "inputTypes": ["old_type"]}'}}
        ]
    }
    codeflash_output = update_edges_with_latest_component_versions(project_data)

def test_empty_project_data():
    project_data = {}
    codeflash_output = update_edges_with_latest_component_versions(project_data)











import json
from collections import defaultdict
from copy import deepcopy

# imports
import pytest  # used for our unit tests
from langflow.initial_setup.setup import \
    update_edges_with_latest_component_versions
# function to test
from loguru import logger

# unit tests



def test_no_edges():
    project_data = {
        "nodes": [
            {"id": "1", "data": {"node": {"outputs": [{"name": "out1", "types": ["type1"]}], "display_name": "SourceNode"}}},
            {"id": "2", "data": {"node": {"template": {"field1": {"input_types": ["type1"]}}, "display_name": "TargetNode"}}}
        ],
        "edges": []
    }
    codeflash_output = update_edges_with_latest_component_versions(project_data)

def test_no_nodes():
    project_data = {
        "nodes": [],
        "edges": [
            {"source": "1", "target": "2", "data": {"sourceHandle": '{"name": "out1"}', "targetHandle": '{"fieldName": "field1"}'}}
        ]
    }
    codeflash_output = update_edges_with_latest_component_versions(project_data)

def test_empty_project_data():
    project_data = {}
    codeflash_output = update_edges_with_latest_component_versions(project_data)







def test_non_matching_source_or_target_nodes():
    project_data = {
        "nodes": [
            {"id": "1", "data": {"node": {"outputs": [{"name": "out1", "types": ["type1"]}], "display_name": "SourceNode"}}}
        ],
        "edges": [
            {"source": "1", "target": "2", "data": {"sourceHandle": '{"name": "out1"}', "targetHandle": '{"fieldName": "field1"}'}}
        ]
    }
    codeflash_output = update_edges_with_latest_component_versions(project_data)

---

📣 **Feedback**

If you have any feedback or need assistance, feel free to join our Discord community: