⚡️ Speed up method JiraDataSource.get_workflow_scheme_draft by 8%

[codeflash-ai]

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

⏱️ Runtime : 2.21 milliseconds → 2.04 milliseconds (best of 49 runs)

📝 Explanation and details

The optimization achieves an 8% runtime improvement through strategic reduction of unnecessary dictionary allocations and operations in the HTTP request preparation path.

Key Optimizations Applied:

1. Conditional Dictionary Creation in get_workflow_scheme_draft():

   • Changed dict(headers or {}) to headers if headers else {} - eliminates unnecessary dict constructor call when headers is already a dict or when no headers are provided
   • Added conditional call to _as_str_dict(_headers) if _headers else {} - avoids string conversion processing for empty headers

2. Early Return in _as_str_dict():

   • Added if not d: return {} check to immediately return empty dict for falsy inputs
   • Eliminates dictionary comprehension execution when input is empty

3. Header Processing Optimization in HTTPClient.execute():

   • Cached request.headers in local variable req_headers to avoid repeated attribute lookups
   • Restructured header merging logic to avoid unnecessary dict copy when request headers are empty
   • Reduced redundant header existence checks

Performance Impact Analysis:

The line profiler shows the most significant gains in:

• get_workflow_scheme_draft(): Reduced from 9.10ms to 8.01ms total time
• _as_str_dict(): Reduced total time from 1.82ms to 1.46ms with fewer calls (1143 → 444 hits)
• HTTPRequest construction time improved due to fewer dictionary operations

Test Case Performance:
The optimizations are particularly effective for:

• Basic cases with no headers (most common scenario) - avoid unnecessary dict allocations
• Concurrent execution scenarios - the per-request overhead reduction compounds across multiple calls
• High-throughput scenarios - reduced object creation pressure benefits sustained load

These micro-optimizations target the hot path of HTTP request preparation, where even small reductions in dictionary operations yield measurable performance gains when executed repeatedly in async contexts.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio

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


# Minimal stubs for HTTPRequest and HTTPResponse to allow tests to run
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, data):
        self.data = data


# Minimal JiraClient and JiraRESTClientViaUsernamePassword for test
class DummyAsyncHTTPClient:
    """Dummy async HTTP client for testing async execute"""

    def __init__(self, base_url):
        self._base_url = base_url
        self.requests = []

    def get_base_url(self):
        return self._base_url

    async def execute(self, req):
        # Simulate an async HTTP call and echo back the request for assertion
        self.requests.append(req)
        # Simulate different responses based on path param id
        if req.path_params.get("id") == "404":
            raise ValueError("Not found")
        return HTTPResponse(
            {
                "method": req.method,
                "url": req.url,
                "headers": req.headers,
                "path_params": req.path_params,
                "query_params": req.query_params,
                "body": req.body,
            }
        )


class DummyJiraClient:
    def __init__(self, base_url):
        self.client = DummyAsyncHTTPClient(base_url)

    def get_client(self):
        return self.client


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


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_basic():
    """Test basic async/await behavior and expected return value."""
    client = DummyJiraClient("https://jira.example.com")
    ds = JiraDataSource(client)
    resp = await ds.get_workflow_scheme_draft(123)


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_with_headers():
    """Test that custom headers are passed and stringified correctly."""
    client = DummyJiraClient("https://jira.example.com/")
    ds = JiraDataSource(client)
    headers = {"X-Test": "abc", "X-Num": 42}
    resp = await ds.get_workflow_scheme_draft(456, headers=headers)


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_id_as_zero():
    """Test with id=0 (edge of valid integer input)."""
    client = DummyJiraClient("https://jira.example.com")
    ds = JiraDataSource(client)
    resp = await ds.get_workflow_scheme_draft(0)


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


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_invalid_client():
    """Test ValueError is raised if HTTP client is not initialized."""

    class BadClient:
        def get_client(self):
            return None

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


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

    class BadHTTPClient:
        pass

    class BadJiraClient:
        def get_client(self):
            return BadHTTPClient()

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


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_concurrent():
    """Test concurrent execution of multiple coroutines."""
    client = DummyJiraClient("https://jira.example.com")
    ds = JiraDataSource(client)
    # Run 5 concurrent requests with different ids
    ids = [101, 102, 103, 104, 105]
    results = await asyncio.gather(*[ds.get_workflow_scheme_draft(i) for i in ids])
    for idx, resp in zip(ids, results):
        pass


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_exception_in_execute():
    """Test that exception in execute is properly propagated."""

    class FailingHTTPClient:
        def get_base_url(self):
            return "https://jira.example.com"

        async def execute(self, req):
            raise RuntimeError("Simulated failure")

    class FailingJiraClient:
        def get_client(self):
            return FailingHTTPClient()

    ds = JiraDataSource(FailingJiraClient())
    with pytest.raises(RuntimeError, match="Simulated failure"):
        await ds.get_workflow_scheme_draft(123)


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_id_as_stringable():
    """Test id as a value that can be stringified (e.g. very large int)."""
    client = DummyJiraClient("https://jira.example.com")
    ds = JiraDataSource(client)
    big_id = 2**62
    resp = await ds.get_workflow_scheme_draft(big_id)


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


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_many_concurrent():
    """Test scalability with a moderate number of concurrent requests."""
    client = DummyJiraClient("https://jira.example.com")
    ds = JiraDataSource(client)
    ids = list(range(20, 70))  # 50 concurrent calls, well below 1000
    coros = [ds.get_workflow_scheme_draft(i) for i in ids]
    results = await asyncio.gather(*coros)
    for idx, resp in zip(ids, results):
        pass


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_headers_and_concurrency():
    """Test concurrency with different headers for each request."""
    client = DummyJiraClient("https://jira.example.com")
    ds = JiraDataSource(client)
    coros = [ds.get_workflow_scheme_draft(i, headers={"X-Req": i}) for i in range(10)]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


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


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_throughput_small_load():
    """Throughput: Test small batch of requests for quick completion."""
    client = DummyJiraClient("https://jira.example.com")
    ds = JiraDataSource(client)
    coros = [ds.get_workflow_scheme_draft(i) for i in range(5)]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_throughput_medium_load():
    """Throughput: Test medium batch of requests for throughput."""
    client = DummyJiraClient("https://jira.example.com")
    ds = JiraDataSource(client)
    coros = [ds.get_workflow_scheme_draft(i) for i in range(50)]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_throughput_varied_headers():
    """Throughput: Test with varied headers and ids."""
    client = DummyJiraClient("https://jira.example.com")
    ds = JiraDataSource(client)
    coros = [
        ds.get_workflow_scheme_draft(i, headers={"X-Index": i, "X-Bool": i % 2 == 0})
        for i in range(30)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_throughput_high_load():
    """Throughput: Test upper-bounded high load (100 concurrent requests)."""
    client = DummyJiraClient("https://jira.example.com")
    ds = JiraDataSource(client)
    coros = [ds.get_workflow_scheme_draft(i) for i in range(100)]
    results = await asyncio.gather(*coros)
    for i, 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.

import asyncio

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

# --- Minimal stubs for required classes and helpers ---


class DummyHTTPResponse:
    """A dummy HTTPResponse for testing."""

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


class DummyHTTPRequest:
    """A dummy HTTPRequest for testing."""

    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


# --- Dummy client classes to simulate JiraDataSource dependencies ---


class DummyClient:
    """A dummy HTTP client for testing async execution."""

    def __init__(self, base_url="https://dummy.atlassian.net"):
        self._base_url = base_url
        self.execute_calls = []
        self.should_raise = None  # If set, raise this exception in execute

    def get_base_url(self):
        return self._base_url

    async def execute(self, req):
        self.execute_calls.append(req)
        if self.should_raise:
            raise self.should_raise
        # Simulate a response with request info for assertion
        return DummyHTTPResponse(
            {
                "method": req.method,
                "url": req.url,
                "headers": req.headers,
                "path_params": req.path_params,
                "query_params": req.query_params,
                "body": req.body,
            }
        )


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

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

    def get_client(self):
        return self._client


# --- TESTS ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_basic_returns_expected_response():
    """Test that the function returns expected values when awaited."""
    dummy_client = DummyClient(base_url="https://jira.example.com/")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)

    response = await ds.get_workflow_scheme_draft(42)


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_basic_with_headers():
    """Test that headers are passed and stringified."""
    dummy_client = DummyClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)

    headers = {"X-Test-Header": "value", "X-Num": 123}
    response = await ds.get_workflow_scheme_draft(99, headers=headers)


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_basic_async_await_behavior():
    """Test that the function is a coroutine and must be awaited."""
    dummy_client = DummyClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)

    codeflash_output = ds.get_workflow_scheme_draft(1)
    coro = codeflash_output
    resp = await coro


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_concurrent_execution():
    """Test concurrent execution of the async function."""
    dummy_client = DummyClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)

    # Run several concurrent requests with different IDs
    ids = [1, 2, 3, 4, 5]
    coros = [ds.get_workflow_scheme_draft(i) for i in ids]
    results = await asyncio.gather(*coros)
    for idx, resp in enumerate(results):
        pass


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

    class NoneClient:
        def get_client(self):
            return None

    jira_client = NoneClient()
    # __init__ should raise ValueError
    with pytest.raises(ValueError, match="HTTP client is not initialized"):
        JiraDataSource(jira_client)


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_raises_if_client_missing_get_base_url():
    """Test that ValueError is raised if client lacks get_base_url."""

    class NoBaseUrlClient:
        def get_client(self):
            class NoBaseUrl:
                pass

            return NoBaseUrl()

    jira_client = 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_workflow_scheme_draft_raises_if_execute_raises():
    """Test that exceptions from the HTTP client are propagated."""
    dummy_client = DummyClient()
    dummy_client.should_raise = RuntimeError("network error")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    with pytest.raises(RuntimeError, match="network error"):
        await ds.get_workflow_scheme_draft(123)


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_headers_and_path_param_types():
    """Test edge cases for header and path param types (e.g., bool, list)."""
    dummy_client = DummyClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    headers = {"X-Bool": True, "X-List": [1, 2, 3]}
    response = await ds.get_workflow_scheme_draft(7, headers=headers)


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_url_formatting_with_missing_param():
    """Test that missing path param does not break URL formatting."""
    dummy_client = DummyClient(base_url="https://jira.example.com/")
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    # The id param is always present, but test that the url is still as expected
    response = await ds.get_workflow_scheme_draft(0)


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_large_scale_concurrent():
    """Test the function with many concurrent calls (scalability)."""
    dummy_client = DummyClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    n = 50  # Reasonable upper bound for unit test speed
    coros = [ds.get_workflow_scheme_draft(i) for i in range(n)]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_large_scale_with_varied_headers():
    """Test the function with many concurrent calls and varied headers."""
    dummy_client = DummyClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    n = 20
    coros = [
        ds.get_workflow_scheme_draft(
            i, headers={"X-Header": f"val-{i}", "X-Flag": bool(i % 2)}
        )
        for i in range(n)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_throughput_small_load():
    """Throughput: Test small batch of requests."""
    dummy_client = DummyClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    coros = [ds.get_workflow_scheme_draft(i) for i in range(5)]
    results = await asyncio.gather(*coros)
    for resp in results:
        pass


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_throughput_medium_load():
    """Throughput: Test medium batch of requests."""
    dummy_client = DummyClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    coros = [ds.get_workflow_scheme_draft(i) for i in range(30)]
    results = await asyncio.gather(*coros)
    for resp in results:
        pass


@pytest.mark.asyncio
async def test_get_workflow_scheme_draft_throughput_sustained_pattern():
    """Throughput: Test sustained execution pattern with staggered awaits."""
    dummy_client = DummyClient()
    jira_client = DummyJiraClient(dummy_client)
    ds = JiraDataSource(jira_client)
    # Launch 10 coroutines, await them in two batches
    coros1 = [ds.get_workflow_scheme_draft(i) for i in range(5)]
    coros2 = [ds.get_workflow_scheme_draft(i) for i in range(5, 10)]
    results1 = await asyncio.gather(*coros1)
    results2 = await asyncio.gather(*coros2)
    for resp in results1 + results2:
        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.get_workflow_scheme_draft-miqbt8ei and push.