⚡️ Speed up function encrypt_auth_settings by 65% in PR #9506 (mcp-composer-integration-v2)

[codeflash-ai]

⚡️ This pull request contains optimizations for PR #9506

If you approve this dependent PR, these changes will be merged into the original PR branch mcp-composer-integration-v2.

This PR will be automatically closed if the original PR is merged.

---

📄 65% (0.65x) speedup for encrypt_auth_settings in langflow/services/auth/mcp_encryption.py

⏱️ Runtime : 7.19 milliseconds → 4.36 milliseconds (best of 29 runs)

📝 Explanation and details

The optimization applies memoization to the get_settings_service() function using @functools.lru_cache(maxsize=1). This is the single key change that delivers the 64% speedup.

What was optimized:

• Added @functools.lru_cache(maxsize=1) decorator to get_settings_service() to cache the expensive service initialization

Why this optimization works:
The line profiler reveals that get_settings_service() was consuming 99.7% of its execution time (58.46ms out of 58.61ms total) in the expensive get_service() call. This function was being called repeatedly - 89 times in the profiled run - but always returning the same SettingsService instance. Each call required:

• Factory initialization via SettingsServiceFactory()
• Service manager lookup and potential registration
• Service instantiation logic

With lru_cache(maxsize=1), the first call performs the expensive initialization, and all subsequent calls return the cached instance in O(1) time. This transforms 89 expensive service lookups into 1 expensive lookup + 88 cache hits.

Performance impact by test type:

• Basic encryption tests: Moderate speedup since they call get_settings_service() fewer times
• Large-scale tests (like test_encrypt_auth_settings_large_number_of_fields): Maximum benefit as they trigger more encryption/decryption operations, each requiring the settings service
• Performance-critical tests: Most significant gains since they stress-test the repeated service access pattern

The optimization maintains identical behavior - same service instance is returned, just cached after first access instead of recreated each time.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	✅ 10 Passed
🌀 Generated Regression Tests	✅ 49 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	90.0%

⚙️ Existing Unit Tests and Runtime

🌀 Generated Regression Tests and Runtime

from __future__ import annotations

import random
import string
from typing import Any

# imports
import pytest  # used for our unit tests
from cryptography.fernet import Fernet, InvalidToken
from langflow.logging.logger import logger
from langflow.services.auth import utils as auth_utils
from langflow.services.auth.mcp_encryption import encrypt_auth_settings
from langflow.services.deps import get_settings_service
from langflow.services.schema import ServiceType
from langflow.services.settings.service import SettingsService

SENSITIVE_FIELDS = [
    "oauth_client_secret",
    "api_key",
]
from langflow.services.auth.mcp_encryption import encrypt_auth_settings

# --- UNIT TESTS ---

# Helper fixture to get a valid SettingsService for encryption/decryption
@pytest.fixture
def settings_service():
    return get_settings_service()

# Helper to generate random strings
def random_string(length):
    return ''.join(random.choices(string.ascii_letters + string.digits, k=length))

# Basic Test Cases

def test_encrypt_auth_settings_none_returns_none():
    """Test that None input returns None."""
    codeflash_output = encrypt_auth_settings(None)

def test_encrypt_auth_settings_no_sensitive_fields(settings_service):
    """Test that dict with no sensitive fields returns unchanged dict."""
    input_dict = {"username": "user", "password": "pass"}
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output

def test_encrypt_auth_settings_empty_sensitive_fields(settings_service):
    """Test that empty sensitive fields are not encrypted."""
    input_dict = {"api_key": "", "oauth_client_secret": None}
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output



def test_encrypt_auth_settings_encrypts_both_fields(settings_service):
    """Test that both sensitive fields are encrypted if present."""
    api_key = "api_key_value"
    secret = "secret_value"
    input_dict = {"api_key": api_key, "oauth_client_secret": secret}
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output


def test_encrypt_auth_settings_preserves_other_fields(settings_service):
    """Test that non-sensitive fields are preserved."""
    input_dict = {
        "api_key": "plain",
        "other_field": "value",
        "oauth_client_secret": "plain_secret",
        "another": 123,
    }
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output

# Edge Test Cases

def test_encrypt_auth_settings_empty_dict(settings_service):
    """Test that empty dict returns empty dict."""
    codeflash_output = encrypt_auth_settings({}); result = codeflash_output





def test_encrypt_auth_settings_sensitive_field_is_empty_string(settings_service):
    """Test that empty string is not encrypted and remains empty."""
    input_dict = {"api_key": ""}
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output

def test_encrypt_auth_settings_sensitive_field_is_whitespace(settings_service):
    """Test that whitespace string is encrypted and can be decrypted."""
    input_dict = {"api_key": " "}
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output

def test_encrypt_auth_settings_sensitive_field_is_unicode(settings_service):
    """Test that unicode string is encrypted and can be decrypted."""
    unicode_str = "pāsswørd😊"
    input_dict = {"api_key": unicode_str}
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output

def test_encrypt_auth_settings_sensitive_field_is_long_string(settings_service):
    """Test that a long string is encrypted and can be decrypted."""
    long_str = random_string(512)
    input_dict = {"oauth_client_secret": long_str}
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output

def test_encrypt_auth_settings_sensitive_field_is_special_characters(settings_service):
    """Test that a string with special characters is encrypted and can be decrypted."""
    special_str = "!@#$%^&*()_+-=[]{}|;':,.<>/?`~"
    input_dict = {"api_key": special_str}
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output

def test_encrypt_auth_settings_sensitive_field_is_multiline(settings_service):
    """Test that a multiline string is encrypted and can be decrypted."""
    multiline = "line1\nline2\nline3"
    input_dict = {"oauth_client_secret": multiline}
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output

# Large Scale Test Cases

def test_encrypt_auth_settings_large_number_of_fields(settings_service):
    """Test scalability with a large number of non-sensitive fields."""
    input_dict = {f"field_{i}": random_string(10) for i in range(1000)}
    input_dict["api_key"] = "large_scale_key"
    input_dict["oauth_client_secret"] = "large_scale_secret"
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output
    # All other fields should be unchanged
    for i in range(1000):
        pass

def test_encrypt_auth_settings_large_sensitive_values(settings_service):
    """Test with very large sensitive values (up to 1000 chars)."""
    large_api_key = random_string(1000)
    large_secret = random_string(1000)
    input_dict = {"api_key": large_api_key, "oauth_client_secret": large_secret}
    codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output

def test_encrypt_auth_settings_many_sensitive_fields(settings_service):
    """Test with many dicts, each with sensitive fields, to check performance."""
    for i in range(20):  # Keep under 1000 for speed
        api_key = random_string(50)
        secret = random_string(50)
        input_dict = {"api_key": api_key, "oauth_client_secret": secret, "other": random_string(10)}
        codeflash_output = encrypt_auth_settings(input_dict); result = codeflash_output

def test_encrypt_auth_settings_idempotency(settings_service):
    """Test that repeated encryption calls do not double-encrypt sensitive fields."""
    api_key = "idempotent_key"
    secret = "idempotent_secret"
    input_dict = {"api_key": api_key, "oauth_client_secret": secret}
    codeflash_output = encrypt_auth_settings(input_dict); once = codeflash_output
    codeflash_output = encrypt_auth_settings(once); twice = codeflash_output
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
from __future__ import annotations

import random
import string
from typing import Any

# imports
import pytest  # used for our unit tests
from cryptography.fernet import Fernet, InvalidToken
from langflow.logging.logger import logger
from langflow.services.auth import utils as auth_utils
from langflow.services.auth.mcp_encryption import encrypt_auth_settings
from langflow.services.deps import get_settings_service
from langflow.services.schema import ServiceType
from langflow.services.settings.service import SettingsService

SENSITIVE_FIELDS = [
    "oauth_client_secret",
    "api_key",
]
from langflow.services.auth.mcp_encryption import encrypt_auth_settings

# --- UNIT TESTS ---

# Helper: Dummy SettingsService with a fixed key
class DummySecret:
    def __init__(self, value):
        self.value = value
    def get_secret_value(self):
        return self.value

class DummyAuthSettings:
    def __init__(self, secret_key):
        self.SECRET_KEY = DummySecret(secret_key)

class DummySettingsService:
    def __init__(self, key="A"*32):
        self.auth_settings = DummyAuthSettings(key)

# --- BASIC TEST CASES ---



def test_encrypt_auth_settings_basic_both_fields():
    # Should encrypt both fields
    auth_settings = {"api_key": "key123", "oauth_client_secret": "secret456"}
    codeflash_output = encrypt_auth_settings(auth_settings); result = codeflash_output

def test_encrypt_auth_settings_basic_no_sensitive_fields():
    # Should leave dict unchanged if no sensitive fields
    auth_settings = {"username": "admin", "password": "pass"}
    codeflash_output = encrypt_auth_settings(auth_settings); result = codeflash_output

def test_encrypt_auth_settings_basic_none_input():
    # Should return None if input is None
    codeflash_output = encrypt_auth_settings(None)

# --- EDGE TEST CASES ---

def test_encrypt_auth_settings_empty_dict():
    # Should return empty dict if input is empty
    codeflash_output = encrypt_auth_settings({}); result = codeflash_output

def test_encrypt_auth_settings_sensitive_field_empty_string():
    # Should not encrypt empty string
    auth_settings = {"api_key": ""}
    codeflash_output = encrypt_auth_settings(auth_settings); result = codeflash_output

def test_encrypt_auth_settings_sensitive_field_none():
    # Should not encrypt None value
    auth_settings = {"api_key": None}
    codeflash_output = encrypt_auth_settings(auth_settings); result = codeflash_output





def test_encrypt_auth_settings_sensitive_field_is_missing():
    # Should not raise if sensitive field is missing
    auth_settings = {"other": "value"}
    codeflash_output = encrypt_auth_settings(auth_settings); result = codeflash_output

def test_encrypt_auth_settings_sensitive_field_is_falsey():
    # Should not encrypt if value is falsey (empty string, None, 0)
    for value in ["", None, 0]:
        auth_settings = {"api_key": value}
        codeflash_output = encrypt_auth_settings(auth_settings); result = codeflash_output




def test_encrypt_auth_settings_large_number_of_fields():
    # Should only encrypt sensitive fields, leave others untouched
    auth_settings = {f"field_{i}": f"value_{i}" for i in range(900)}
    auth_settings["api_key"] = "large_key"
    auth_settings["oauth_client_secret"] = "large_secret"
    codeflash_output = encrypt_auth_settings(auth_settings); result = codeflash_output
    # All non-sensitive fields should be unchanged
    for i in range(900):
        pass

def test_encrypt_auth_settings_many_sensitive_fields():
    # Should encrypt all sensitive fields present
    auth_settings = {"api_key": "key", "oauth_client_secret": "secret"}
    for i in range(998):
        auth_settings[f"api_key_{i}"] = f"val_{i}"
    codeflash_output = encrypt_auth_settings(auth_settings); result = codeflash_output
    # Other fields should remain unchanged
    for i in range(998):
        pass


def test_encrypt_auth_settings_performance_large_dict():
    # Performance: should not take excessive time for large dicts
    import time
    auth_settings = {f"field_{i}": f"value_{i}" for i in range(999)}
    auth_settings["api_key"] = "key"
    start = time.time()
    codeflash_output = encrypt_auth_settings(auth_settings); result = codeflash_output
    duration = time.time() - start

To edit these changes git checkout codeflash/optimize-pr9506-2025-09-04T16.16.29 and push.

[coderabbitai]

Important

Review skipped

Bot user detected.

To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

---

🪧 Tips

Chat

There are 3 ways to chat with CodeRabbit:

• Review comments: Directly reply to a review comment made by CodeRabbit. Example:
  • I pushed a fix in commit <commit_id>, please review it.
  • Open a follow-up GitHub issue for this discussion.
• Files and specific lines of code (under the "Files changed" tab): Tag @coderabbitai in a new review comment at the desired location with your query.
• PR comments: Tag @coderabbitai in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
  • @coderabbitai gather interesting stats about this repository and render them as a table. Additionally, render a pie chart showing the language distribution in the codebase.
  • @coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.

Support

Need help? Join our Discord community for assistance with any issues or questions.

CodeRabbit Commands (Invoked using PR/Issue comments)

Type @coderabbitai help to get the list of available commands.

Other keywords and placeholders

• Add @coderabbitai ignore or @coderabbit ignore anywhere in the PR description to prevent this PR from being reviewed.
• Add @coderabbitai summary to generate the high-level summary at a specific location in the PR description.
• Add @coderabbitai anywhere in the PR title to generate the title automatically.

Status, Documentation and Community

• Visit our Status Page to check the current availability of CodeRabbit.
• Visit our Documentation for detailed information on how to use CodeRabbit.
• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[codeflash-ai]

This PR has been automatically closed because the original PR #9506 by jordanrfrazier was closed.