⚡️ Speed up method JiraDataSource.get_configuration by 28%

[codeflash-ai]

📄 28% (0.28x) speedup for JiraDataSource.get_configuration in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 2.72 milliseconds → 2.12 milliseconds (best of 250 runs)

📝 Explanation and details

The optimized code achieves a 28% runtime improvement and 12.6% throughput increase through a targeted micro-optimization in the _as_str_dict() helper function.

Key Optimization:
The optimization adds an early return check for empty dictionaries in _as_str_dict():

def _as_str_dict(d: Dict[str, Any]) -> Dict[str, str]:
    if not d:  # Early return for empty dicts
        return {}
    return {str(k): _serialize_value(v) for k, v in d.items()}

Why This Provides Speedup:

1. Eliminates unnecessary dictionary comprehensions - In the original code, empty dictionaries still triggered the comprehension {str(k): _serialize_value(v) for k, v in d.items()}, even though d.items() would be empty
2. Reduces function call overhead - The early return avoids calling _serialize_value() and str() conversions when there's no work to do
3. Memory allocation savings - Skips creating a new dictionary through comprehension when returning an empty dict literal is sufficient

Performance Impact Analysis:
From the line profiler results, _as_str_dict() is called 1,893 times during the test run. The optimization shows:

• Original: 2.0989ms total time across all calls
• Optimized: 1.76258ms total time across all calls
• This represents a 16% reduction in time spent in this helper function alone

The function is called multiple times per get_configuration() request (for headers, path_params, and query_params), making this optimization particularly effective for high-throughput scenarios where many requests contain empty parameter dictionaries.

Test Case Benefits:
The optimization is most beneficial for test cases with empty or minimal parameters, which is common in API calls like get_configuration() that often have empty path parameters and query parameters. The concurrent execution tests (50-200 calls) see amplified benefits due to the multiplicative effect of the optimization across many requests.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
# Import the classes and helpers from the provided code
from typing import Any, Dict, Optional

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


# Minimal HTTPResponse stub for testing
class HTTPResponse:
    def __init__(self, response_data):
        self._data = response_data

    def json(self):
        return self._data

    def __eq__(self, other):
        if not isinstance(other, HTTPResponse):
            return False
        return self._data == other._data

# Minimal HTTPRequest stub for testing
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

# Minimal JiraRESTClientViaToken stub for testing
class JiraRESTClientViaToken:
    def __init__(self, base_url: str, token: str = "dummy", token_type: str = "Bearer"):
        self.base_url = base_url

    def get_base_url(self) -> str:
        return self.base_url

    async def execute(self, request: HTTPRequest):
        # Simulate a response based on request for testing
        # For edge/throughput tests, allow custom response injection via headers
        resp_data = {
            "method": request.method,
            "url": request.url,
            "headers": request.headers,
            "path_params": request.path_params,
            "query_params": request.query_params,
            "body": request.body,
        }
        # Simulate error if requested
        if request.headers.get("X-Force-Error") == "true":
            raise RuntimeError("Simulated error")
        # Simulate large payload if requested
        if request.headers.get("X-Large-Response") == "true":
            resp_data["large"] = ["item"] * int(request.headers.get("X-Size", "100"))
        return HTTPResponse(resp_data)

# Minimal JiraClient stub for testing
class JiraClient:
    def __init__(self, client):
        self.client = client

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

# ------------------ TESTS ------------------

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_get_configuration_returns_expected_response():
    """Test that get_configuration returns an HTTPResponse with correct method and URL."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    resp = await datasource.get_configuration()
    # Check response contents
    resp_json = resp.json()

@pytest.mark.asyncio
async def test_get_configuration_with_custom_headers():
    """Test get_configuration with custom headers."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    custom_headers = {"Authorization": "Bearer testtoken", "X-Test": "123"}
    resp = await datasource.get_configuration(headers=custom_headers)
    resp_json = resp.json()

@pytest.mark.asyncio
async def test_get_configuration_async_await_behavior():
    """Test that get_configuration is a coroutine and can be awaited."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    # Should return a coroutine before awaiting
    codeflash_output = datasource.get_configuration(); coro = codeflash_output
    resp = await coro

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_get_configuration_concurrent_execution():
    """Test concurrent execution of get_configuration."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    # Run 10 concurrent requests
    results = await asyncio.gather(
        *[datasource.get_configuration(headers={"X-Test": str(i)}) for i in range(10)]
    )
    # All results should be HTTPResponse and have unique headers
    for idx, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_get_configuration_exception_handling():
    """Test that get_configuration propagates exceptions from execute."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    # Pass a header to trigger error in stub
    with pytest.raises(RuntimeError, match="Simulated error"):
        await datasource.get_configuration(headers={"X-Force-Error": "true"})

@pytest.mark.asyncio
async def test_get_configuration_invalid_client_raises():
    """Test that JiraDataSource raises if client is None or lacks get_base_url."""
    # Client is None
    class DummyClient:
        def get_client(self):
            return None
    with pytest.raises(ValueError, match="HTTP client is not initialized"):
        JiraDataSource(DummyClient())
    # Client lacks get_base_url
    class BadClient:
        def get_client(self):
            return object()
    with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
        JiraDataSource(BadClient())

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_get_configuration_large_scale_concurrent():
    """Test large scale concurrent execution (up to 100 calls)."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    num_calls = 100
    tasks = [
        datasource.get_configuration(headers={"X-Test": f"bulk-{i}"})
        for i in range(num_calls)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_get_configuration_large_payload_response():
    """Test get_configuration handles large responses."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    large_size = 500  # Not exceeding 1000
    resp = await datasource.get_configuration(headers={"X-Large-Response": "true", "X-Size": str(large_size)})
    resp_json = resp.json()

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_get_configuration_throughput_small_load():
    """Throughput: Test small load of 5 concurrent calls."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    num_calls = 5
    tasks = [datasource.get_configuration(headers={"X-Test": f"tp-{i}"}) for i in range(num_calls)]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_get_configuration_throughput_medium_load():
    """Throughput: Test medium load of 50 concurrent calls."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    num_calls = 50
    tasks = [datasource.get_configuration(headers={"X-Test": f"tp-{i}"}) for i in range(num_calls)]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_get_configuration_throughput_high_volume():
    """Throughput: Test high volume load of 200 concurrent calls."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    num_calls = 200
    tasks = [datasource.get_configuration(headers={"X-Test": f"tp-{i}"}) for i in range(num_calls)]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_get_configuration_throughput_large_payload():
    """Throughput: Test large payload response under concurrent load."""
    base_url = "https://jira.example.com"
    client = JiraClient(JiraRESTClientViaToken(base_url))
    datasource = JiraDataSource(client)
    num_calls = 10
    large_size = 250
    tasks = [
        datasource.get_configuration(headers={"X-Large-Response": "true", "X-Size": str(large_size)})
        for _ in range(num_calls)
    ]
    results = await asyncio.gather(*tasks)
    for resp in results:
        resp_json = resp.json()
# 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
from typing import Any, Dict, Union

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 minimal dummy HTTPResponse to simulate real HTTPResponse objects."""
    def __init__(self, data):
        self.data = data

class DummyAsyncClient:
    """A dummy async HTTP client to simulate execute method."""
    def __init__(self):
        self.executed_requests = []
        self.should_raise = False
        self.response_data = "ok"

    async def execute(self, request):
        """Simulate async execution of a request."""
        self.executed_requests.append(request)
        if self.should_raise:
            raise RuntimeError("Simulated execution error")
        return DummyHTTPResponse(self.response_data)

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

class DummyJiraClient:
    """A dummy JiraClient for testing JiraDataSource."""
    def __init__(self, client=None):
        self._client = client

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

# --- Minimal HTTPRequest stub for request creation ---

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

# --- UNIT TESTS FOR JiraDataSource.get_configuration ---

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_get_configuration_basic_returns_response():
    """Test that get_configuration returns a DummyHTTPResponse with expected data."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    resp = await ds.get_configuration()

@pytest.mark.asyncio
async def test_get_configuration_with_custom_headers():
    """Test that custom headers are passed and formatted correctly."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    custom_headers = {"X-Test": "123", "Another": 456}
    resp = await ds.get_configuration(headers=custom_headers)
    # The DummyAsyncClient stores the last request for inspection
    last_req = dummy_client.executed_requests[-1]

@pytest.mark.asyncio
async def test_get_configuration_basic_async_await_behavior():
    """Ensure the function is a coroutine and requires await."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    codeflash_output = ds.get_configuration(); coro = codeflash_output
    resp = await coro

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_get_configuration_client_none_raises():
    """Test that ValueError is raised if client.get_client() returns None."""
    jira_client = DummyJiraClient(None)
    with pytest.raises(ValueError, match="HTTP client is not initialized"):
        JiraDataSource(jira_client)

@pytest.mark.asyncio
async def test_get_configuration_client_missing_get_base_url():
    """Test ValueError if client lacks get_base_url method."""
    class NoBaseUrlClient:
        pass
    jira_client = DummyJiraClient(NoBaseUrlClient())
    with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
        JiraDataSource(jira_client)

@pytest.mark.asyncio
async def test_get_configuration_execute_raises_exception():
    """Test that exceptions during execute are propagated."""
    dummy_client = DummyAsyncClient()
    dummy_client.should_raise = True
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    with pytest.raises(RuntimeError, match="Simulated execution error"):
        await ds.get_configuration()

@pytest.mark.asyncio
async def test_get_configuration_concurrent_execution():
    """Test concurrent calls to get_configuration (should all succeed)."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    # Run 5 concurrent requests
    results = await asyncio.gather(
        *(ds.get_configuration() for _ in range(5))
    )

@pytest.mark.asyncio
async def test_get_configuration_empty_headers_dict():
    """Test that passing an empty headers dict works and produces empty headers."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    resp = await ds.get_configuration(headers={})
    last_req = dummy_client.executed_requests[-1]

@pytest.mark.asyncio
async def test_get_configuration_headers_with_various_types():
    """Test that headers with various value types are stringified."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    headers = {"int": 1, "bool": True, "float": 3.14, "none": None, "list": [1, 2]}
    await ds.get_configuration(headers=headers)
    last_req = dummy_client.executed_requests[-1]

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_get_configuration_many_concurrent_calls():
    """Test many concurrent get_configuration calls for scalability."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    num_calls = 50  # Reasonable, not excessive
    results = await asyncio.gather(*(ds.get_configuration() for _ in range(num_calls)))

@pytest.mark.asyncio
async def test_get_configuration_concurrent_mixed_headers():
    """Test concurrent calls with different headers."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    headers_list = [{"X-Req": str(i)} for i in range(10)]
    results = await asyncio.gather(*(ds.get_configuration(headers=h) for h in headers_list))
    for idx, req in enumerate(dummy_client.executed_requests[-10:]):
        pass

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_get_configuration_throughput_small_load():
    """Throughput: small batch of requests."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    results = await asyncio.gather(*(ds.get_configuration() for _ in range(5)))

@pytest.mark.asyncio
async def test_get_configuration_throughput_medium_load():
    """Throughput: medium batch of requests."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    results = await asyncio.gather(*(ds.get_configuration() for _ in range(40)))

@pytest.mark.asyncio
async def test_get_configuration_throughput_large_load():
    """Throughput: large batch of requests, but still bounded."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    results = await asyncio.gather(*(ds.get_configuration() for _ in range(100)))

@pytest.mark.asyncio
async def test_get_configuration_throughput_sustained_pattern():
    """Throughput: sustained pattern of sequential and concurrent calls."""
    dummy_client = DummyAsyncClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    # Sequential calls
    for _ in range(10):
        resp = await ds.get_configuration()
    # Concurrent burst
    results = await asyncio.gather(*(ds.get_configuration() for _ in range(20)))
    # Final sequential
    for _ in range(5):
        resp = await ds.get_configuration()
# 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.get_configuration-mhp3d74j and push.