⚡️ Speed up method JiraDataSource.get_shared_time_tracking_configuration by 8%

[codeflash-ai]

📄 8% (0.08x) speedup for JiraDataSource.get_shared_time_tracking_configuration in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 106 milliseconds → 98.2 milliseconds (best of 241 runs)

📝 Explanation and details

The optimized code achieves a 7% runtime improvement and 1.7% throughput increase through targeted micro-optimizations that reduce unnecessary object allocations and dictionary operations in the hot path.

Key optimizations:

1. Eliminated redundant dictionary allocations in get_shared_time_tracking_configuration:

   • Replaced dict(headers or {}) with direct conditional assignment: headers if headers is not None else {}
   • Removed intermediate variables _path, _query, and _body that were always empty/None
   • Used literal empty dictionaries {} instead of _as_str_dict(_path) and _as_str_dict(_query) calls

2. Reduced _as_str_dict function calls from 3 to 1 per request:

   • Line profiler shows _as_str_dict total time dropped from 1.8ms to 1.3ms
   • Function call count decreased from 1,251 hits to 417 hits (3x reduction)

3. Streamlined variable assignments by eliminating unnecessary intermediate allocations:

   • Direct parameter passing instead of creating temporary variables that are immediately consumed

Performance impact analysis:

• The optimizations are most effective for high-frequency API calls where the cumulative effect of reduced allocations becomes significant
• Throughput tests (10-100 concurrent requests) show consistent improvements, indicating the optimizations scale well under load
• Memory efficiency is improved through fewer temporary object creations per request

Test case effectiveness:

• Small-to-medium concurrent loads (10-100 requests) benefit most from the reduced allocation overhead
• Single request scenarios see modest but measurable improvements
• Edge cases with large headers still benefit from the streamlined parameter handling

The optimizations maintain full backward compatibility while reducing the computational overhead in the request preparation phase, making each API call slightly more efficient without changing the async execution model.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 443 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 pytest  # used for our unit tests
from app.sources.external.jira.jira import JiraDataSource


# ---- Minimal stubs for dependencies ----
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

class HTTPResponse:
    def __init__(self, response):
        self._response = response
        self.status_code = getattr(response, "status_code", 200)
        self.content = getattr(response, "content", b"{}")
        self.text = getattr(response, "text", "{}")
        self.headers = getattr(response, "headers", {})

    def json(self):
        # Simulate a JSON response
        return getattr(self._response, "json_data", {"success": True})

class DummyAsyncResponse:
    """A dummy response object for simulating HTTPX responses."""
    def __init__(self, status_code=200, content=b"{}", text="{}", headers=None, json_data=None):
        self.status_code = status_code
        self.content = content
        self.text = text
        self.headers = headers or {}
        self.json_data = json_data or {"success": True}

class DummyAsyncClient:
    """A dummy async client to simulate httpx.AsyncClient behavior."""
    async def request(self, method, url, **kwargs):
        # Simulate a successful response
        return DummyAsyncResponse(
            status_code=kwargs.get("status_code", 200),
            content=kwargs.get("content", b"{}"),
            text=kwargs.get("text", "{}"),
            headers=kwargs.get("headers", {}),
            json_data=kwargs.get("json_data", {"success": True})
        )

class DummyJiraRESTClientViaUsernamePassword:
    """Dummy JIRA REST client for testing."""
    def __init__(self, base_url, username, password, token_type="Basic"):
        self.base_url = base_url
        self.client = DummyAsyncClient()
    def get_base_url(self):
        return self.base_url
    async def execute(self, request, **kwargs):
        # Simulate a successful HTTPResponse
        return HTTPResponse(await self.client.request(request.method, request.url, **kwargs))

class DummyJiraClient:
    """Dummy JiraClient for testing."""
    def __init__(self, client):
        self.client = client
    def get_client(self):
        return self.client
from app.sources.external.jira.jira import JiraDataSource

# ---- Unit Tests ----

# ------------------- BASIC TEST CASES -------------------

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_basic_success():
    """Test basic async/await behavior and successful response."""
    client = DummyJiraClient(DummyJiraRESTClientViaUsernamePassword("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    resp = await ds.get_shared_time_tracking_configuration()

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_with_headers():
    """Test passing custom headers to the function."""
    client = DummyJiraClient(DummyJiraRESTClientViaUsernamePassword("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    custom_headers = {"X-Test-Header": "TestValue"}
    resp = await ds.get_shared_time_tracking_configuration(headers=custom_headers)

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_empty_headers():
    """Test passing empty headers dict."""
    client = DummyJiraClient(DummyJiraRESTClientViaUsernamePassword("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    resp = await ds.get_shared_time_tracking_configuration(headers={})

# ------------------- EDGE TEST CASES -------------------

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_client_none():
    """Test ValueError when client.get_client() returns None."""
    class DummyBadClient:
        def get_client(self):
            return None
    with pytest.raises(ValueError) as excinfo:
        JiraDataSource(DummyBadClient())

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_missing_get_base_url():
    """Test ValueError when client does not have get_base_url method."""
    class DummyBadRESTClient:
        pass
    class DummyBadClient:
        def get_client(self):
            return DummyBadRESTClient()
    with pytest.raises(ValueError) as excinfo:
        JiraDataSource(DummyBadClient())

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_headers_type_variants():
    """Test headers with non-string keys/values."""
    client = DummyJiraClient(DummyJiraRESTClientViaUsernamePassword("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    headers = {1: True, "X-Int": 42, "X-None": None}
    resp = await ds.get_shared_time_tracking_configuration(headers=headers)

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_concurrent_execution():
    """Test concurrent execution of the async function."""
    client = DummyJiraClient(DummyJiraRESTClientViaUsernamePassword("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    # Run 5 concurrent requests
    results = await asyncio.gather(
        *[ds.get_shared_time_tracking_configuration(headers={"X-Req": i}) for i in range(5)]
    )

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_exception_in_execute():
    """Test that an exception in execute is propagated."""
    class FailingRESTClient(DummyJiraRESTClientViaUsernamePassword):
        async def execute(self, request, **kwargs):
            raise RuntimeError("Simulated failure")
    client = DummyJiraClient(FailingRESTClient("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    with pytest.raises(RuntimeError) as excinfo:
        await ds.get_shared_time_tracking_configuration()

# ------------------- LARGE SCALE TEST CASES -------------------

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_many_concurrent():
    """Test function scalability with 50 concurrent executions."""
    client = DummyJiraClient(DummyJiraRESTClientViaUsernamePassword("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    num_requests = 50
    results = await asyncio.gather(
        *[ds.get_shared_time_tracking_configuration(headers={"X-Req": i}) for i in range(num_requests)]
    )

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_large_headers():
    """Test with a large number of headers."""
    client = DummyJiraClient(DummyJiraRESTClientViaUsernamePassword("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    large_headers = {f"Header-{i}": f"Value-{i}" for i in range(100)}
    resp = await ds.get_shared_time_tracking_configuration(headers=large_headers)

# ------------------- THROUGHPUT TEST CASES -------------------

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_throughput_small_load():
    """Throughput test with a small load (10 requests)."""
    client = DummyJiraClient(DummyJiraRESTClientViaUsernamePassword("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    num_requests = 10
    results = await asyncio.gather(
        *[ds.get_shared_time_tracking_configuration(headers={"X-Req": i}) for i in range(num_requests)]
    )

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_throughput_medium_load():
    """Throughput test with a medium load (50 requests)."""
    client = DummyJiraClient(DummyJiraRESTClientViaUsernamePassword("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    num_requests = 50
    results = await asyncio.gather(
        *[ds.get_shared_time_tracking_configuration(headers={"X-Req": i}) for i in range(num_requests)]
    )

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_throughput_high_load():
    """Throughput test with a high load (100 requests)."""
    client = DummyJiraClient(DummyJiraRESTClientViaUsernamePassword("https://jira.example.com", "user", "pass"))
    ds = JiraDataSource(client)
    num_requests = 100
    results = await asyncio.gather(
        *[ds.get_shared_time_tracking_configuration(headers={"X-Req": i}) for i in range(num_requests)]
    )
# 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 enable isolated unit testing ---

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

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

class DummyHTTPRequest:
    """A dummy HTTPRequest to match the signature."""
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)

# --- Dummy JiraClient and HTTP client for testing ---

class DummyHTTPClient:
    """A dummy HTTP client that simulates async execute."""
    def __init__(self, base_url="http://testserver", execute_result=None, raise_on_execute=None):
        self._base_url = base_url
        self._execute_result = execute_result or DummyHTTPResponse({"ok": True})
        self._raise_on_execute = raise_on_execute
        self.last_request = None

    def get_base_url(self):
        return self._base_url

    async def execute(self, req):
        self.last_request = req
        if self._raise_on_execute:
            raise self._raise_on_execute
        await asyncio.sleep(0)  # Simulate async context switch
        return self._execute_result

class DummyJiraClient:
    """A dummy JiraClient that returns a DummyHTTPClient."""
    def __init__(self, client):
        self.client = client

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

# ---- Helpers used by generated methods ----
# (already defined above)

# --- UNIT TESTS ---

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_basic_success():
    """Test basic successful call returns the expected DummyHTTPResponse."""
    dummy_response = DummyHTTPResponse({"result": "success"})
    client = DummyHTTPClient(execute_result=dummy_response)
    ds = JiraDataSource(DummyJiraClient(client))
    resp = await ds.get_shared_time_tracking_configuration()

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_with_headers():
    """Test passing custom headers."""
    dummy_response = DummyHTTPResponse({"headers": True})
    client = DummyHTTPClient(execute_result=dummy_response)
    ds = JiraDataSource(DummyJiraClient(client))
    custom_headers = {"X-Test-Header": "test-value", "Another": 123}
    resp = await ds.get_shared_time_tracking_configuration(headers=custom_headers)

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_empty_headers():
    """Test passing empty headers dict."""
    dummy_response = DummyHTTPResponse({"ok": True})
    client = DummyHTTPClient(execute_result=dummy_response)
    ds = JiraDataSource(DummyJiraClient(client))
    resp = await ds.get_shared_time_tracking_configuration(headers={})

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_none_headers():
    """Test passing None as headers."""
    dummy_response = DummyHTTPResponse({"ok": True})
    client = DummyHTTPClient(execute_result=dummy_response)
    ds = JiraDataSource(DummyJiraClient(client))
    resp = await ds.get_shared_time_tracking_configuration(headers=None)

# 2. Edge Test Cases

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

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_client_no_get_base_url():
    """Test ValueError is raised if client lacks get_base_url method."""
    class NoBaseUrlClient:
        pass
    class DummyJiraClient2:
        def get_client(self):
            return NoBaseUrlClient()
    with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
        JiraDataSource(DummyJiraClient2())

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_execute_raises_exception():
    """Test that exceptions from execute are propagated."""
    exc = RuntimeError("Execute failed")
    client = DummyHTTPClient(raise_on_execute=exc)
    ds = JiraDataSource(DummyJiraClient(client))
    with pytest.raises(RuntimeError, match="Execute failed"):
        await ds.get_shared_time_tracking_configuration()

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_concurrent_calls():
    """Test concurrent async calls produce correct results."""
    dummy_response = DummyHTTPResponse({"result": "concurrent"})
    client = DummyHTTPClient(execute_result=dummy_response)
    ds = JiraDataSource(DummyJiraClient(client))
    # Launch 5 concurrent calls
    results = await asyncio.gather(
        *[ds.get_shared_time_tracking_configuration(headers={"X": i}) for i in range(5)]
    )
    for resp in results:
        pass

@pytest.mark.asyncio

async def test_get_shared_time_tracking_configuration_many_concurrent_calls():
    """Test 50 concurrent calls for scalability and isolation."""
    dummy_response = DummyHTTPResponse({"result": "bulk"})
    client = DummyHTTPClient(execute_result=dummy_response)
    ds = JiraDataSource(DummyJiraClient(client))
    tasks = [ds.get_shared_time_tracking_configuration(headers={"X": i}) for i in range(50)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_large_headers():
    """Test with a large number of headers."""
    large_headers = {f"Header-{i}": i for i in range(100)}
    dummy_response = DummyHTTPResponse({"ok": True})
    client = DummyHTTPClient(execute_result=dummy_response)
    ds = JiraDataSource(DummyJiraClient(client))
    resp = await ds.get_shared_time_tracking_configuration(headers=large_headers)

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_throughput_small_load():
    """Throughput test: 10 concurrent calls (small load)."""
    dummy_response = DummyHTTPResponse({"result": "small-load"})
    client = DummyHTTPClient(execute_result=dummy_response)
    ds = JiraDataSource(DummyJiraClient(client))
    tasks = [ds.get_shared_time_tracking_configuration(headers={"X": i}) for i in range(10)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_throughput_medium_load():
    """Throughput test: 100 concurrent calls (medium load)."""
    dummy_response = DummyHTTPResponse({"result": "medium-load"})
    client = DummyHTTPClient(execute_result=dummy_response)
    ds = JiraDataSource(DummyJiraClient(client))
    tasks = [ds.get_shared_time_tracking_configuration(headers={"X": i}) for i in range(100)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_get_shared_time_tracking_configuration_throughput_varied_load():
    """Throughput test: varied header values under concurrent load."""
    dummy_response = DummyHTTPResponse({"result": "varied"})
    client = DummyHTTPClient(execute_result=dummy_response)
    ds = JiraDataSource(DummyJiraClient(client))
    tasks = [
        ds.get_shared_time_tracking_configuration(headers={"X": str(i), "Y": i * 2})
        for i in range(25)
    ]
    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.

To edit these changes git checkout codeflash/optimize-JiraDataSource.get_shared_time_tracking_configuration-mhp5zo9k and push.