⚡️ Speed up method JiraDataSource.update_workflow_scheme_mappings by 7%

[codeflash-ai]

📄 7% (0.07x) speedup for JiraDataSource.update_workflow_scheme_mappings in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 1.88 milliseconds → 1.75 milliseconds (best of 52 runs)

📝 Explanation and details

The optimized code achieves a 7% speedup through strategic elimination of unnecessary dictionary operations in hot paths, despite showing a slight throughput decrease due to measurement variance.

Key Optimizations Applied:

1. Smart headers handling: Changed dict(headers or {}) to dict(headers) if headers else {} - this avoids creating an intermediate empty dict when headers is None, reducing allocation overhead.

2. Eliminated redundant empty dict allocations: Removed creation of _path and _query variables that were always empty, directly passing {} literals to HTTPRequest constructor and _safe_format_url.

3. Fast-path optimization in _as_str_dict: Added early return if not d: return {} to skip the expensive dict comprehension entirely for empty dictionaries - this is particularly effective since the profiler shows this function being called 873 times in the original vs 291 times in the optimized version.

Performance Impact Analysis:

• The line profiler shows the most expensive operations were HTTPRequest construction (55.6% → 46.2% of time) and _as_str_dict calls, both of which were optimized
• _as_str_dict execution time dropped from 1.23ms to 0.93ms (25% improvement)
• Total function time reduced from 7.54ms to 6.43ms

Test Case Benefits:
The optimizations are most effective for scenarios with:

• Empty or minimal headers (most test cases)
• High-frequency API calls (throughput tests with 5-100 concurrent requests)
• Standard workflow operations without custom path/query parameters

The slight throughput decrease (-1.9%) appears to be measurement noise, as the core runtime improvement of 7% demonstrates the optimization's effectiveness in reducing per-call overhead in this Jira API integration hot path.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
from typing import Any, Dict

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

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


class HTTPResponse:
    """A simple HTTPResponse stub for testing."""

    def __init__(self, data: Any = None, status_code: int = 200):
        self.data = data
        self.status_code = status_code


class HTTPRequest:
    """A simple HTTPRequest stub for testing."""

    def __init__(
        self,
        method: str,
        url: str,
        headers: Dict[str, str],
        path_params: Dict[str, str],
        query_params: Dict[str, str],
        body: Any,
    ):
        self.method = method
        self.url = url
        self.headers = headers
        self.path_params = path_params
        self.query_params = query_params
        self.body = body


# --- Minimal JiraClient stub for testing ---
class DummyJiraRESTClient:
    """Dummy Jira REST client with async execute and get_base_url."""

    def __init__(self, base_url: str):
        self.base_url = base_url
        self.last_request = None
        self.should_raise = False
        self.response_data = None

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

    async def execute(self, request: HTTPRequest):
        self.last_request = request
        if self.should_raise:
            raise RuntimeError("Simulated execute failure")
        # Simulate a response, echoing back the request body for validation
        return HTTPResponse(data={"echo": request.body}, status_code=200)


class DummyJiraClient:
    """Dummy JiraClient that returns DummyJiraRESTClient."""

    def __init__(self, client: DummyJiraRESTClient):
        self.client = client

    def get_client(self):
        return self.client


# --- Decorator stub (no-op) ---
def codeflash_capture(*args, **kwargs):
    def decorator(fn):
        return fn

    return decorator


# --- TESTS START HERE ---

# 1. BASIC TEST CASES


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_basic():
    """Test basic async/await and correct request construction."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    workflows = [{"issueType": "Bug", "workflowId": "wf1"}]
    resp = await ds.update_workflow_scheme_mappings("scheme123", workflows)


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_with_default_workflow():
    """Test passing defaultWorkflowId and header override."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com/")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    workflows = [{"issueType": "Task", "workflowId": "wf2"}]
    headers = {"X-Custom-Header": "value"}
    resp = await ds.update_workflow_scheme_mappings(
        "scheme456", workflows, defaultWorkflowId="wf-default", headers=headers
    )
    # Check that Content-Type header is present and custom header is set
    sent_headers = dummy_client.last_request.headers


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_empty_workflows():
    """Test with empty workflowsForIssueTypes list."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    workflows = []
    resp = await ds.update_workflow_scheme_mappings("scheme789", workflows)


# 2. EDGE TEST CASES


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_missing_client():
    """Test ValueError is raised if client is None."""

    class NullClient:
        def get_client(self):
            return None

    with pytest.raises(ValueError, match="HTTP client is not initialized"):
        JiraDataSource(NullClient())


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_client_missing_get_base_url():
    """Test ValueError if client lacks get_base_url."""

    class IncompleteClient:
        def get_client(self):
            return object()

    with pytest.raises(
        ValueError, match="HTTP client does not have get_base_url method"
    ):
        JiraDataSource(IncompleteClient())


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_execute_raises():
    """Test if execute raises an exception, it propagates."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com")
    dummy_client.should_raise = True
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    with pytest.raises(RuntimeError, match="Simulated execute failure"):
        await ds.update_workflow_scheme_mappings(
            "schemeX", [{"issueType": "Epic", "workflowId": "wfE"}]
        )


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_concurrent_calls():
    """Test concurrent execution of multiple calls."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    workflows1 = [{"issueType": "Bug", "workflowId": "wf1"}]
    workflows2 = [{"issueType": "Task", "workflowId": "wf2"}]
    # Run two calls concurrently
    results = await asyncio.gather(
        ds.update_workflow_scheme_mappings("scheme1", workflows1),
        ds.update_workflow_scheme_mappings(
            "scheme2", workflows2, defaultWorkflowId="wfD"
        ),
    )


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_headers_are_copied():
    """Test that the headers dict is copied and not mutated."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    headers = {"X-Test": "A"}
    await ds.update_workflow_scheme_mappings(
        "scheme", [{"issueType": "Bug", "workflowId": "wf"}], headers=headers
    )


# 3. LARGE SCALE TEST CASES


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_large_workflows():
    """Test with a large number of workflow mappings (but < 1000)."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    workflows = [{"issueType": f"Issue{i}", "workflowId": f"wf{i}"} for i in range(300)]
    resp = await ds.update_workflow_scheme_mappings("schemeLarge", workflows)


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_many_concurrent():
    """Test many concurrent calls (up to 50, well under 1000)."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    workflows = [{"issueType": "Bug", "workflowId": "wf"}]
    coros = [
        ds.update_workflow_scheme_mappings(f"scheme{i}", workflows) for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    for i, r in enumerate(results):
        pass


# 4. THROUGHPUT TEST CASES


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_throughput_small_load():
    """Throughput test: small load (5 concurrent requests)."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    workflows = [{"issueType": "Bug", "workflowId": "wf"}]
    coros = [
        ds.update_workflow_scheme_mappings(f"scheme{i}", workflows) for i in range(5)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_throughput_medium_load():
    """Throughput test: medium load (20 concurrent requests)."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    workflows = [{"issueType": "Task", "workflowId": "wf"}]
    coros = [
        ds.update_workflow_scheme_mappings(f"scheme{i}", workflows) for i in range(20)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_throughput_large_load():
    """Throughput test: large load (100 concurrent requests, < 1000)."""
    dummy_client = DummyJiraRESTClient("https://jira.example.com")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    workflows = [{"issueType": "Epic", "workflowId": "wf"}]
    coros = [
        ds.update_workflow_scheme_mappings(f"scheme{i}", workflows) for i in range(100)
    ]
    results = await asyncio.gather(*coros)


# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

import asyncio
# Import the function to test (copy-paste from the provided code)

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


# Mocks and helpers for testing
class DummyAsyncClient:
    """A dummy async HTTP client for testing."""

    def __init__(self, response=None, raise_exc=None):
        self.response = response
        self.raise_exc = raise_exc
        self.last_request = None

    async def request(self, method, url, **kwargs):
        self.last_request = (method, url, kwargs)
        if self.raise_exc:
            raise self.raise_exc
        # Simulate an httpx.Response-like object
        return DummyHTTPResponse(
            self.response
            or {"status": "ok", "url": url, "method": method, "kwargs": kwargs}
        )


class DummyHTTPResponse:
    """A dummy HTTP response for testing."""

    def __init__(self, data):
        self.data = data

    def json(self):
        return self.data


class DummyHTTPClient:
    """A dummy HTTPClient that mimics the real one for async tests."""

    def __init__(
        self, base_url="https://jira.example.com", response=None, raise_exc=None
    ):
        self.base_url = base_url
        self._client = DummyAsyncClient(response=response, raise_exc=raise_exc)

    def get_base_url(self):
        return self.base_url

    async def execute(self, request, **kwargs):
        resp = await self._client.request(request.method, request.url, **kwargs)
        return DummyHTTPResponse(resp.data)


class DummyJiraClient:
    """A dummy JiraClient wrapper."""

    def __init__(self, client):
        self.client = client

    def get_client(self):
        return self.client


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


# ---- Basic Test Cases ----


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_basic_success():
    """Test basic async call returns expected DummyHTTPResponse."""
    dummy_client = DummyJiraClient(DummyHTTPClient())
    ds = JiraDataSource(dummy_client)
    workflows = [{"issueType": "Bug", "workflowId": "wf1"}]
    resp = await ds.update_workflow_scheme_mappings(
        id="scheme1", workflowsForIssueTypes=workflows
    )


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_with_default_workflow_id():
    """Test passing defaultWorkflowId is reflected in request body."""
    dummy_client = DummyJiraClient(DummyHTTPClient())
    ds = JiraDataSource(dummy_client)
    workflows = [{"issueType": "Task", "workflowId": "wf2"}]
    resp = await ds.update_workflow_scheme_mappings(
        id="scheme2", workflowsForIssueTypes=workflows, defaultWorkflowId="defaultWF"
    )
    # Check the body in the last request
    last_body = dummy_client.get_client()._client.last_request[2].get("json")


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_empty_workflows():
    """Test with empty workflowsForIssueTypes (edge case)."""
    dummy_client = DummyJiraClient(DummyHTTPClient())
    ds = JiraDataSource(dummy_client)
    resp = await ds.update_workflow_scheme_mappings(
        id="scheme4", workflowsForIssueTypes=[]
    )
    last_body = dummy_client.get_client()._client.last_request[2].get("json")


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_none_client_raises():
    """Test ValueError is raised if client is None."""

    class BrokenJiraClient:
        def get_client(self):
            return None

    with pytest.raises(ValueError, match="HTTP client is not initialized"):
        JiraDataSource(BrokenJiraClient())


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_client_missing_base_url():
    """Test ValueError if client lacks get_base_url method."""

    class NoBaseUrlClient:
        pass

    class Wrapper:
        def get_client(self):
            return NoBaseUrlClient()

    with pytest.raises(
        ValueError, match="HTTP client does not have get_base_url method"
    ):
        JiraDataSource(Wrapper())


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_execute_raises_exception():
    """Test exception in execute is propagated."""
    dummy_client = DummyJiraClient(DummyHTTPClient(raise_exc=RuntimeError("fail")))
    ds = JiraDataSource(dummy_client)
    workflows = [{"issueType": "Bug", "workflowId": "wf1"}]
    with pytest.raises(RuntimeError, match="fail"):
        await ds.update_workflow_scheme_mappings(
            id="scheme5", workflowsForIssueTypes=workflows
        )


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_concurrent_execution():
    """Test concurrent async calls work independently."""
    dummy_client = DummyJiraClient(DummyHTTPClient())
    ds = JiraDataSource(dummy_client)
    workflows1 = [{"issueType": "Bug", "workflowId": "wf1"}]
    workflows2 = [{"issueType": "Task", "workflowId": "wf2"}]
    results = await asyncio.gather(
        ds.update_workflow_scheme_mappings(
            id="schemeA", workflowsForIssueTypes=workflows1
        ),
        ds.update_workflow_scheme_mappings(
            id="schemeB", workflowsForIssueTypes=workflows2
        ),
    )


# ---- Large Scale Test Cases ----


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_large_workflows():
    """Test with a large number of workflowsForIssueTypes."""
    dummy_client = DummyJiraClient(DummyHTTPClient())
    ds = JiraDataSource(dummy_client)
    workflows = [{"issueType": f"Type{i}", "workflowId": f"wf{i}"} for i in range(100)]
    resp = await ds.update_workflow_scheme_mappings(
        id="schemeLarge", workflowsForIssueTypes=workflows
    )
    last_body = dummy_client.get_client()._client.last_request[2].get("json")


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_many_concurrent_calls():
    """Test many concurrent async calls for scalability."""
    dummy_client = DummyJiraClient(DummyHTTPClient())
    ds = JiraDataSource(dummy_client)
    workflows = [{"issueType": "Bug", "workflowId": "wf1"}]
    tasks = [
        ds.update_workflow_scheme_mappings(
            id=f"scheme{i}", workflowsForIssueTypes=workflows
        )
        for i in range(20)
    ]
    results = await asyncio.gather(*tasks)


# ---- Throughput Test Cases ----


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_throughput_small_load():
    """Throughput: small load of concurrent requests."""
    dummy_client = DummyJiraClient(DummyHTTPClient())
    ds = JiraDataSource(dummy_client)
    workflows = [{"issueType": "Bug", "workflowId": "wf1"}]
    tasks = [
        ds.update_workflow_scheme_mappings(
            id=f"scheme{i}", workflowsForIssueTypes=workflows
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_throughput_medium_load():
    """Throughput: medium load of concurrent requests."""
    dummy_client = DummyJiraClient(DummyHTTPClient())
    ds = JiraDataSource(dummy_client)
    workflows = [{"issueType": "Task", "workflowId": "wf2"}]
    tasks = [
        ds.update_workflow_scheme_mappings(
            id=f"scheme{i}", workflowsForIssueTypes=workflows
        )
        for i in range(25)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_update_workflow_scheme_mappings_throughput_large_load():
    """Throughput: large load of concurrent requests (stress test, <100)."""
    dummy_client = DummyJiraClient(DummyHTTPClient())
    ds = JiraDataSource(dummy_client)
    workflows = [{"issueType": "Epic", "workflowId": "wf3"}]
    tasks = [
        ds.update_workflow_scheme_mappings(
            id=f"scheme{i}", workflowsForIssueTypes=workflows
        )
        for i in range(50)
    ]
    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.update_workflow_scheme_mappings-miq5wcrm and push.