⚡️ Speed up method BitwardenService._get_sensitive_information_from_identity by 9%

[codeflash-ai]

📄 9% (0.09x) speedup for BitwardenService._get_sensitive_information_from_identity in skyvern/forge/sdk/services/bitwarden.py

⏱️ Runtime : 7.65 milliseconds → 7.00 milliseconds (best of 13 runs)

📝 Explanation and details

The optimized code achieves a 9% runtime improvement and 8.3% throughput improvement through two key optimizations targeting the most expensive operations:

1. Environment Copy Reduction in run_command
The original code called os.environ.copy() on every subprocess invocation (754 hits consuming 24.9% of total time). The optimization caches a base environment copy as a class attribute and only performs additional copies when additional_env is provided. This reduces the expensive environment copying from every call to just once per class lifecycle, plus selective copies only when needed.

2. Field Lookup Algorithm Optimization in _get_sensitive_information_from_identity
The original code used nested loops for field extraction - for each requested field, it linearly searched through all custom fields (for item in identity_item["fields"]). With 17,947 hits on the inner loop, this created O(NM) complexity where N = number of requested fields and M = number of custom fields. The optimization pre-builds a lookup dictionary (custom_fields_lookup) converting field searches to O(1) operations, reducing the algorithm from O(NM) to O(N+M).

Performance Impact Analysis:
Based on the function reference, _get_sensitive_information_from_identity is called through a retry wrapper that can invoke it multiple times with exponential timeout backoff. The run_command method is heavily used across login, sync, unlock, list items, and logout operations. These optimizations are particularly beneficial for:

• Large-scale concurrent operations (as shown in the 100-concurrent-call test case)
• High-frequency Bitwarden operations where subprocess creation overhead accumulates
• Items with many custom fields where the O(N*M) search penalty was most severe

The test results confirm the optimization maintains correctness across all scenarios while delivering consistent performance gains, especially valuable given this code runs in automated workflows that may perform many Bitwarden operations sequentially or concurrently.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
import json
import os
from unittest.mock import AsyncMock, MagicMock, patch

import pytest  # used for our unit tests
from skyvern.config import settings
from skyvern.exceptions import (BitwardenAccessDeniedError,
                                BitwardenListItemsError, BitwardenLoginError,
                                BitwardenLogoutError, BitwardenSyncError,
                                BitwardenUnlockError)
from skyvern.forge.sdk.services.bitwarden import BitwardenService

class RunCommandResult:
    """Dummy RunCommandResult for test context."""
    def __init__(self, stdout, stderr, returncode=0):
        self.stdout = stdout
        self.stderr = stderr
        self.returncode = returncode

# ----------------- UNIT TESTS -----------------

@pytest.mark.asyncio
async def test_get_sensitive_information_basic_custom_field():
    """Test basic extraction from custom fields."""
    # Patch all external async dependencies to return controlled results
    with patch.object(BitwardenService, "login", new=AsyncMock()), \
         patch.object(BitwardenService, "sync", new=AsyncMock()), \
         patch.object(BitwardenService, "unlock", new=AsyncMock(return_value="SESSION123")), \
         patch.object(BitwardenService, "logout", new=AsyncMock()), \
         patch.object(BitwardenService, "run_command", new=AsyncMock()) as mock_run:

        # Simulate Bitwarden CLI returning a single item with custom fields
        item = {
            "fields": [{"name": "secret1", "value": "value1"}, {"name": "secret2", "value": "value2"}],
            "identity": {"email": "test@example.com"},
        }
        mock_run.return_value = RunCommandResult(stdout=json.dumps([item]), stderr="", returncode=0)

        result = await BitwardenService._get_sensitive_information_from_identity(
            client_id="id", client_secret="sec", master_password="pw",
            collection_id="colid", identity_key="key",
            identity_fields=["secret1", "secret2"], bw_organization_id=None, bw_collection_ids=None
        )

@pytest.mark.asyncio
async def test_get_sensitive_information_basic_identity_field():
    """Test extraction from identity sub-dict when not in custom fields."""
    with patch.object(BitwardenService, "login", new=AsyncMock()), \
         patch.object(BitwardenService, "sync", new=AsyncMock()), \
         patch.object(BitwardenService, "unlock", new=AsyncMock(return_value="SESSION123")), \
         patch.object(BitwardenService, "logout", new=AsyncMock()), \
         patch.object(BitwardenService, "run_command", new=AsyncMock()) as mock_run:

        # Custom fields do not contain the field, but identity does
        item = {
            "fields": [],
            "identity": {"email": "test@example.com", "phone": "12345"},
        }
        mock_run.return_value = RunCommandResult(stdout=json.dumps([item]), stderr="", returncode=0)

        result = await BitwardenService._get_sensitive_information_from_identity(
            client_id="id", client_secret="sec", master_password="pw",
            collection_id="colid", identity_key="key",
            identity_fields=["email", "phone"], bw_organization_id=None, bw_collection_ids=None
        )

@pytest.mark.asyncio

async def test_get_sensitive_information_empty_fields():
    """Test with empty identity_fields returns empty dict."""
    with patch.object(BitwardenService, "login", new=AsyncMock()), \
         patch.object(BitwardenService, "sync", new=AsyncMock()), \
         patch.object(BitwardenService, "unlock", new=AsyncMock(return_value="SESSION123")), \
         patch.object(BitwardenService, "logout", new=AsyncMock()), \
         patch.object(BitwardenService, "run_command", new=AsyncMock()) as mock_run:

        item = {
            "fields": [{"name": "foo", "value": "bar"}],
            "identity": {"baz": "qux"},
        }
        mock_run.return_value = RunCommandResult(stdout=json.dumps([item]), stderr="", returncode=0)

        result = await BitwardenService._get_sensitive_information_from_identity(
            client_id="id", client_secret="sec", master_password="pw",
            collection_id="colid", identity_key="key",
            identity_fields=[], bw_organization_id=None, bw_collection_ids=None
        )

@pytest.mark.asyncio

async def test_get_sensitive_information_invalid_json():
    """Test that invalid JSON in run_command output raises BitwardenListItemsError."""
    with patch.object(BitwardenService, "login", new=AsyncMock()), \
         patch.object(BitwardenService, "sync", new=AsyncMock()), \
         patch.object(BitwardenService, "unlock", new=AsyncMock(return_value="SESSION123")), \
         patch.object(BitwardenService, "logout", new=AsyncMock()), \
         patch.object(BitwardenService, "run_command", new=AsyncMock()) as mock_run:
        mock_run.return_value = RunCommandResult(stdout="not a json", stderr="", returncode=0)
        with pytest.raises(BitwardenListItemsError):
            await BitwardenService._get_sensitive_information_from_identity(
                client_id="id", client_secret="sec", master_password="pw",
                collection_id="colid", identity_key="key",
                identity_fields=["foo"], bw_organization_id=None, bw_collection_ids=None
            )

@pytest.mark.asyncio
async def test_get_sensitive_information_no_items():
    """Test that empty items list raises BitwardenListItemsError."""
    with patch.object(BitwardenService, "login", new=AsyncMock()), \
         patch.object(BitwardenService, "sync", new=AsyncMock()), \
         patch.object(BitwardenService, "unlock", new=AsyncMock(return_value="SESSION123")), \
         patch.object(BitwardenService, "logout", new=AsyncMock()), \
         patch.object(BitwardenService, "run_command", new=AsyncMock()) as mock_run:
        mock_run.return_value = RunCommandResult(stdout="[]", stderr="", returncode=0)
        with pytest.raises(BitwardenListItemsError):
            await BitwardenService._get_sensitive_information_from_identity(
                client_id="id", client_secret="sec", master_password="pw",
                collection_id="colid", identity_key="key",
                identity_fields=["foo"], bw_organization_id=None, bw_collection_ids=None
            )

@pytest.mark.asyncio

async def test_get_sensitive_information_concurrent():
    """Test concurrent calls to the function do not interfere."""
    with patch.object(BitwardenService, "login", new=AsyncMock()), \
         patch.object(BitwardenService, "sync", new=AsyncMock()), \
         patch.object(BitwardenService, "unlock", new=AsyncMock(return_value="SESSION123")), \
         patch.object(BitwardenService, "logout", new=AsyncMock()), \
         patch.object(BitwardenService, "run_command", new=AsyncMock()) as mock_run:

        item = {
            "fields": [{"name": "foo", "value": "bar"}],
            "identity": {},
        }
        mock_run.return_value = RunCommandResult(stdout=json.dumps([item]), stderr="", returncode=0)

        async def call_func(val):
            res = await BitwardenService._get_sensitive_information_from_identity(
                client_id="id", client_secret="sec", master_password="pw",
                collection_id="colid", identity_key="key",
                identity_fields=["foo"], bw_organization_id=None, bw_collection_ids=None
            )
            return res

        # Run 5 concurrent calls
        results = await asyncio.gather(*(call_func(i) for i in range(5)))
        for r in results:
            pass

@pytest.mark.asyncio
async def test_get_sensitive_information_logout_always_called():
    """Test that logout is always called, even if an error occurs."""
    with patch.object(BitwardenService, "login", new=AsyncMock()), \
         patch.object(BitwardenService, "sync", new=AsyncMock()), \
         patch.object(BitwardenService, "unlock", new=AsyncMock(return_value="SESSION123")), \
         patch.object(BitwardenService, "logout", new=AsyncMock()) as mock_logout, \
         patch.object(BitwardenService, "run_command", new=AsyncMock()) as mock_run:

        mock_run.return_value = RunCommandResult(stdout="not a json", stderr="", returncode=0)
        with pytest.raises(BitwardenListItemsError):
            await BitwardenService._get_sensitive_information_from_identity(
                client_id="id", client_secret="sec", master_password="pw",
                collection_id="colid", identity_key="key",
                identity_fields=["foo"], bw_organization_id=None, bw_collection_ids=None
            )

@pytest.mark.asyncio
async def test_get_sensitive_information_large_scale():
    """Test with a large number of identity fields and custom fields."""
    with patch.object(BitwardenService, "login", new=AsyncMock()), \
         patch.object(BitwardenService, "sync", new=AsyncMock()), \
         patch.object(BitwardenService, "unlock", new=AsyncMock(return_value="SESSION123")), \
         patch.object(BitwardenService, "logout", new=AsyncMock()), \
         patch.object(BitwardenService, "run_command", new=AsyncMock()) as mock_run:

        # 100 custom fields
        fields = [{"name": f"field{i}", "value": f"value{i}"} for i in range(100)]
        item = {"fields": fields, "identity": {}}
        mock_run.return_value = RunCommandResult(stdout=json.dumps([item]), stderr="", returncode=0)

        identity_fields = [f"field{i}" for i in range(100)]
        result = await BitwardenService._get_sensitive_information_from_identity(
            client_id="id", client_secret="sec", master_password="pw",
            collection_id="colid", identity_key="key",
            identity_fields=identity_fields, bw_organization_id=None, bw_collection_ids=None
        )

@pytest.mark.asyncio

async def test_get_sensitive_information_throughput_large_load():
    """Throughput test: large load of 100 concurrent calls."""
    with patch.object(BitwardenService, "login", new=AsyncMock()), \
         patch.object(BitwardenService, "sync", new=AsyncMock()), \
         patch.object(BitwardenService, "unlock", new=AsyncMock(return_value="SESSION123")), \
         patch.object(BitwardenService, "logout", new=AsyncMock()), \
         patch.object(BitwardenService, "run_command", new=AsyncMock()) as mock_run:

        item = {
            "fields": [{"name": "foo", "value": "bar"}],
            "identity": {},
        }
        mock_run.return_value = RunCommandResult(stdout=json.dumps([item]), stderr="", returncode=0)

        async def call_func():
            return await BitwardenService._get_sensitive_information_from_identity(
                client_id="id", client_secret="sec", master_password="pw",
                collection_id="colid", identity_key="key",
                identity_fields=["foo"], bw_organization_id=None, bw_collection_ids=None
            )
        results = await asyncio.gather(*(call_func() for _ in range(100)))
        for r in results:
            pass
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To edit these changes git checkout codeflash/optimize-BitwardenService._get_sensitive_information_from_identity-mirhdzmc and push.