⚡️ Speed up method JiraDataSource.bulk_edit_dashboards by 15%

[codeflash-ai]

📄 15% (0.15x) speedup for JiraDataSource.bulk_edit_dashboards in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 2.27 milliseconds → 1.97 milliseconds (best of 250 runs)

📝 Explanation and details

The optimization achieves a 15% runtime improvement and 5% throughput improvement by consolidating dictionary initialization in the bulk_edit_dashboards method.

Key optimization: Instead of creating an empty _body dictionary and then adding required fields with separate assignments:

_body: Dict[str, Any] = {}
_body['action'] = action
_body['entityIds'] = entityIds

The optimized version initializes the dictionary with required fields directly:

_body: Dict[str, Any] = {
    'action': action,
    'entityIds': entityIds,
}

Why this improves performance:

• Eliminates redundant dictionary operations: The original code performed 3 separate dictionary operations (empty dict creation + 2 key assignments), while the optimized version does this in a single dictionary literal creation
• Reduces Python bytecode overhead: Dictionary literals are more efficient than multiple __setitem__ calls in Python's bytecode execution
• Better memory access patterns: Single allocation vs. multiple hash table insertions

Impact analysis:
The line profiler shows the optimization reduces time spent on body construction from ~490ns total (235ns + 265ns + 239ns) to ~424ns total (252ns + 171ns + 154ns). While this seems small per call, the throughput improvement of 5% demonstrates meaningful gains when processing many requests concurrently.

Test case performance: The optimization benefits all test patterns equally since every call constructs the request body. Concurrent test cases (like test_bulk_edit_dashboards_throughput_medium_load with 100 concurrent calls) particularly benefit from the reduced per-request overhead, as the optimization compounds across all concurrent operations.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio
from typing import Any, Dict, Optional, Union

import pytest
from app.sources.external.jira.jira import JiraDataSource

# --- Minimal stubs for dependencies to allow isolated testing ---

class DummyHTTPResponse:
    """A simple dummy HTTPResponse for testing."""
    def __init__(self, data):
        self.data = data

    def __eq__(self, other):
        # Allow direct comparison in asserts
        if isinstance(other, DummyHTTPResponse):
            return self.data == other.data
        return False

class DummyAsyncClient:
    """A dummy async client that records requests and returns a canned response."""
    def __init__(self):
        self.requests = []
        self.should_raise = None
        self.response = DummyHTTPResponse({'ok': True})

    async def execute(self, request):
        if self.should_raise:
            raise self.should_raise
        self.requests.append(request)
        return self.response

    def get_base_url(self):
        return "https://dummy.atlassian.net"

class DummyJiraClient:
    """A wrapper to provide get_client() as expected by JiraDataSource."""
    def __init__(self, client):
        self.client = client

    def get_client(self):
        return self.client
from app.sources.external.jira.jira import JiraDataSource


class HTTPRequest:
    def __init__(self, method, url, headers, path_params, query_params, body):
        self.method = method
        self.url = url
        self.headers = headers
        self.path_params = path_params
        self.query_params = query_params
        self.body = body

# --- TESTS ---

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_basic_success():
    """Test that the function returns the dummy HTTPResponse for minimal valid input."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    resp = await ds.bulk_edit_dashboards(
        action="grant",
        entityIds=[1]
    )
    req = dummy_client.requests[0]

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_all_optional_fields():
    """Test that all optional fields are passed correctly in the body."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    resp = await ds.bulk_edit_dashboards(
        action="revoke",
        entityIds=[2, 3],
        changeOwnerDetails={"owner": "user1"},
        extendAdminPermissions=True,
        permissionDetails={"perm": "edit"},
        headers={"X-Custom": "foo"}
    )
    req = dummy_client.requests[0]

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_content_type_override():
    """Test that Content-Type is not overridden if provided in headers."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    await ds.bulk_edit_dashboards(
        action="grant",
        entityIds=[1],
        headers={"Content-Type": "application/x-custom"}
    )
    req = dummy_client.requests[0]

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_empty_entityIds():
    """Test with empty entityIds list (should still call client)."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    resp = await ds.bulk_edit_dashboards(
        action="grant",
        entityIds=[]
    )
    req = dummy_client.requests[0]

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_none_client_raises():
    """Test that ValueError is raised if client is None."""
    class DummyJiraClientNone:
        def get_client(self):
            return None
    with pytest.raises(ValueError, match="HTTP client is not initialized"):
        JiraDataSource(DummyJiraClientNone())

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_client_missing_get_base_url():
    """Test that ValueError is raised if client lacks get_base_url()."""
    class ClientWithoutBaseUrl:
        pass
    with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
        JiraDataSource(DummyJiraClient(ClientWithoutBaseUrl()))

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_client_execute_exception():
    """Test that exceptions raised by client.execute are propagated."""
    dummy_client = DummyAsyncClient()
    dummy_client.should_raise = RuntimeError("fail!")
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    with pytest.raises(RuntimeError, match="fail!"):
        await ds.bulk_edit_dashboards(
            action="grant",
            entityIds=[1]
        )

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_concurrent_calls():
    """Test concurrent execution of multiple bulk_edit_dashboards calls."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    # Launch 5 concurrent calls with different entityIds
    tasks = [
        ds.bulk_edit_dashboards(
            action="grant",
            entityIds=[i]
        ) for i in range(5)
    ]
    results = await asyncio.gather(*tasks)
    # Each request should have unique entityIds
    ids_seen = set()
    for req in dummy_client.requests:
        ids_seen.add(tuple(req.body['entityIds']))

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_large_entityIds():
    """Test with a large list of entityIds."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    entity_ids = list(range(100))
    resp = await ds.bulk_edit_dashboards(
        action="grant",
        entityIds=entity_ids
    )
    req = dummy_client.requests[0]

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_many_concurrent_calls():
    """Test many concurrent calls to bulk_edit_dashboards."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    n = 50  # Keep well below 1000 for speed
    tasks = [
        ds.bulk_edit_dashboards(
            action="grant",
            entityIds=[i, i+1]
        ) for i in range(n)
    ]
    results = await asyncio.gather(*tasks)

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_throughput_small_load():
    """Throughput: Test 10 concurrent calls with small payloads."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    tasks = [
        ds.bulk_edit_dashboards(
            action="grant",
            entityIds=[i]
        ) for i in range(10)
    ]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_throughput_medium_load():
    """Throughput: Test 100 concurrent calls with medium payloads."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    tasks = [
        ds.bulk_edit_dashboards(
            action="grant",
            entityIds=[i, i+1, i+2]
        ) for i in range(100)
    ]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_throughput_large_payload():
    """Throughput: Test single call with large entityIds payload."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    entity_ids = list(range(500))
    resp = await ds.bulk_edit_dashboards(
        action="grant",
        entityIds=entity_ids
    )
    req = dummy_client.requests[0]

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_throughput_sustained_pattern():
    """Throughput: Test repeated calls in a sustained pattern."""
    dummy_client = DummyAsyncClient()
    ds = JiraDataSource(DummyJiraClient(dummy_client))
    # Simulate 5 rounds of 20 concurrent calls
    for round in range(5):
        tasks = [
            ds.bulk_edit_dashboards(
                action="grant",
                entityIds=[round, i]
            ) for i in range(20)
        ]
        results = await asyncio.gather(*tasks)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import asyncio  # used to run async functions

import pytest  # used for our unit tests
from app.sources.external.jira.jira import JiraDataSource

# --- Minimal stubs for dependencies to allow isolated testing ---

class DummyHTTPResponse:
    """A dummy HTTPResponse to simulate real HTTPResponse objects."""
    def __init__(self, data):
        self.data = data

    def __eq__(self, other):
        # For assertion purposes in tests
        return isinstance(other, DummyHTTPResponse) and self.data == other.data

class DummyAsyncClient:
    """A dummy async client that mimics the interface of the real HTTP client."""
    def __init__(self, base_url, should_fail=False, delay=0):
        self._base_url = base_url
        self.should_fail = should_fail
        self.delay = delay
        self.last_request = None

    def get_base_url(self):
        return self._base_url

    async def execute(self, req):
        # Optionally simulate a delay
        if self.delay > 0:
            await asyncio.sleep(self.delay)
        self.last_request = req
        if self.should_fail:
            raise RuntimeError("Simulated HTTP client failure")
        # Return a dummy response with the request data for inspection
        return DummyHTTPResponse({"method": req.method, "url": req.url, "body": req.body, "headers": req.headers})

class DummyJiraClient:
    """A dummy JiraClient wrapper for the underlying HTTP client."""
    def __init__(self, client):
        self.client = client

    def get_client(self):
        return self.client

# --- Minimal HTTPRequest stub for request construction ---

class HTTPRequest:
    def __init__(self, method, url, headers, path_params, query_params, body):
        self.method = method
        self.url = url
        self.headers = headers
        self.path_params = path_params
        self.query_params = query_params
        self.body = body
from app.sources.external.jira.jira import JiraDataSource

# --- Unit Tests ---

# 1. BASIC TEST CASES

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_basic_minimal():
    """Test basic async/await call with minimal required arguments."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    result = await ds.bulk_edit_dashboards(action="grant", entityIds=[1, 2, 3])

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_basic_all_fields():
    """Test all optional and required fields provided."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    result = await ds.bulk_edit_dashboards(
        action="revoke",
        entityIds=[42],
        changeOwnerDetails={"newOwner": "alice"},
        extendAdminPermissions=True,
        permissionDetails={"role": "admin"},
        headers={"Authorization": "Bearer testtoken"}
    )

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_basic_content_type_override():
    """Test that custom Content-Type header is respected."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    result = await ds.bulk_edit_dashboards(
        action="grant",
        entityIds=[1],
        headers={"Content-Type": "application/custom"}
    )

# 2. EDGE TEST CASES

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_empty_entity_ids():
    """Test with empty entityIds list (edge case)."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    result = await ds.bulk_edit_dashboards(action="grant", entityIds=[])

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_none_client_raises():
    """Test that ValueError is raised if client is None."""
    class DummyJiraClientNone:
        def get_client(self):
            return None
    with pytest.raises(ValueError, match="HTTP client is not initialized"):
        JiraDataSource(DummyJiraClientNone())

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_missing_get_base_url_raises():
    """Test that ValueError is raised if get_base_url is missing."""
    class ClientNoBaseUrl:
        pass
    class DummyJiraClientBad:
        def get_client(self):
            return ClientNoBaseUrl()
    with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
        JiraDataSource(DummyJiraClientBad())

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_client_execute_exception():
    """Test that exceptions from the underlying client are propagated."""
    client = DummyAsyncClient(base_url="https://jira.example.com", should_fail=True)
    ds = JiraDataSource(DummyJiraClient(client))
    with pytest.raises(RuntimeError, match="Simulated HTTP client failure"):
        await ds.bulk_edit_dashboards(action="grant", entityIds=[1, 2])

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_concurrent_execution():
    """Test concurrent execution of the async function."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    # Launch multiple concurrent calls with different arguments
    results = await asyncio.gather(
        ds.bulk_edit_dashboards(action="grant", entityIds=[1]),
        ds.bulk_edit_dashboards(action="revoke", entityIds=[2, 3], extendAdminPermissions=False),
        ds.bulk_edit_dashboards(action="grant", entityIds=[4, 5, 6], changeOwnerDetails={"newOwner": "bob"}),
    )

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_entity_ids_types():
    """Test entityIds with various integer values (edge values)."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    # Use negative, zero, and large int
    ids = [-1, 0, 2**31-1]
    result = await ds.bulk_edit_dashboards(action="grant", entityIds=ids)

# 3. LARGE SCALE TEST CASES

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_large_entity_ids():
    """Test with a large number of entityIds."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    large_ids = list(range(1000))  # 1000 entity IDs
    result = await ds.bulk_edit_dashboards(action="grant", entityIds=large_ids)

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_concurrent_large_scale():
    """Test many concurrent executions (scalability)."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    tasks = [
        ds.bulk_edit_dashboards(action="grant", entityIds=[i])
        for i in range(20)
    ]
    results = await asyncio.gather(*tasks)
    for idx, resp in enumerate(results):
        pass

# 4. THROUGHPUT TEST CASES

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_throughput_small_load():
    """Throughput: Test with a small batch of concurrent requests."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    tasks = [
        ds.bulk_edit_dashboards(action="grant", entityIds=[i, i+1])
        for i in range(10)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_throughput_medium_load():
    """Throughput: Test with a medium batch of concurrent requests."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    tasks = [
        ds.bulk_edit_dashboards(action="grant", entityIds=[i])
        for i in range(50)
    ]
    results = await asyncio.gather(*tasks)
    for idx, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_bulk_edit_dashboards_throughput_high_volume():
    """Throughput: Test with a high volume of concurrent requests (but <1000)."""
    client = DummyAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(DummyJiraClient(client))
    tasks = [
        ds.bulk_edit_dashboards(action="grant", entityIds=[i])
        for i in range(100)
    ]
    results = await asyncio.gather(*tasks)
    for idx, resp in enumerate(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-JiraDataSource.bulk_edit_dashboards-mhp8kenu and push.