⚡️ Speed up method JiraDataSource.get_default_workflow by 28%

[codeflash-ai]

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

⏱️ Runtime : 4.36 milliseconds → 3.41 milliseconds (best of 17 runs)

📝 Explanation and details

The optimized code achieves a 27% runtime improvement through several key optimizations that reduce function call overhead and unnecessary dictionary conversions:

Key Optimizations:

1. Conditional header processing: Only calls _as_str_dict(headers) when headers are actually present and non-empty, avoiding unnecessary conversions for the common case of empty headers (451/775 calls had empty headers).

2. Direct path parameter construction: Instead of creating {'id': id} and then converting via _as_str_dict(), the code directly serializes the ID and constructs the string dictionary, eliminating one function call per request.

3. Conditional query parameter handling: Only creates and populates the query dictionary when returnDraftIfExists is not None, avoiding empty dict processing.

4. Simplified _safe_format_url: Removes the _SafeDict wrapper since the path parameters are now pre-converted to strings, reducing object allocation overhead.

Performance Impact:

The line profiler shows the most significant improvements in:

• _as_str_dict calls reduced from 2,325 to 324 (86% reduction)
• HTTPRequest creation time reduced from 11.27ms to 2.39ms (79% improvement)
• Overall function time reduced from 18.16ms to 12.63ms

Test Case Benefits:

These optimizations are particularly effective for:

• High-throughput scenarios with many concurrent requests (200+ requests)
• Cases with minimal or no custom headers (most common usage pattern)
• Batch processing where the same optimization pattern repeats frequently

The optimizations maintain identical functionality while significantly reducing CPU overhead for HTTP request preparation in JIRA API calls.

✅ Correctness verification report:

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

🌀 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

# Mocks and minimal implementations for required classes


class HTTPResponse:
    """Minimal HTTPResponse mock for testing."""

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


class HTTPRequest:
    """Minimal HTTPRequest mock 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


class MockAsyncClient:
    """Mock async client to simulate HTTPClient behavior."""

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

    def get_base_url(self):
        return self.base_url

    async def execute(self, req):
        # Simulate a successful response with request info for validation
        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 JiraClient:
    """Mock JiraClient builder."""

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

    def get_client(self):
        return self.client


# ------------------- UNIT TESTS -------------------

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_get_default_workflow_basic_id_only():
    """Test basic usage with only required id argument."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    resp = await ds.get_default_workflow(42)


@pytest.mark.asyncio
async def test_get_default_workflow_with_returnDraftIfExists_true():
    """Test usage with returnDraftIfExists=True."""
    client = MockAsyncClient(base_url="https://jira.example.com/")
    ds = JiraDataSource(JiraClient(client))
    resp = await ds.get_default_workflow(99, returnDraftIfExists=True)


@pytest.mark.asyncio
async def test_get_default_workflow_with_headers():
    """Test passing custom headers."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    resp = await ds.get_default_workflow(1, headers={"X-Test": "abc"})


# 2. Edge Test Cases


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

    class BadClient:
        pass

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


@pytest.mark.asyncio
async def test_get_default_workflow_returnDraftIfExists_false():
    """Test returnDraftIfExists=False is serialized correctly."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    resp = await ds.get_default_workflow(7, returnDraftIfExists=False)


@pytest.mark.asyncio
async def test_get_default_workflow_headers_and_query():
    """Test headers and query params together."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    resp = await ds.get_default_workflow(
        123, returnDraftIfExists=True, headers={"Authorization": "Bearer xyz"}
    )


@pytest.mark.asyncio
async def test_get_default_workflow_concurrent_execution():
    """Test concurrent execution of multiple requests."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    ids = [1, 2, 3, 4, 5]
    results = await asyncio.gather(*(ds.get_default_workflow(i) for i in ids))
    # Each response should have correct path param
    for i, resp in zip(ids, results):
        pass


@pytest.mark.asyncio
async def test_get_default_workflow_id_as_zero():
    """Edge case: id=0 should be formatted correctly."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    resp = await ds.get_default_workflow(0)


@pytest.mark.asyncio
async def test_get_default_workflow_negative_id():
    """Edge case: negative id should be formatted correctly."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    resp = await ds.get_default_workflow(-123)


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_get_default_workflow_many_concurrent_requests():
    """Test large scale concurrent execution (50 requests)."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    ids = list(range(50))
    results = await asyncio.gather(*(ds.get_default_workflow(i) for i in ids))
    # Validate each response
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_get_default_workflow_large_headers():
    """Test with large number of headers (100)."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    headers = {f"Header-{i}": f"Value-{i}" for i in range(100)}
    resp = await ds.get_default_workflow(5, headers=headers)
    # Validate all headers are present
    for i in range(100):
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_get_default_workflow_throughput_small_load():
    """Throughput: small load (10 requests)."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    ids = list(range(10))
    results = await asyncio.gather(*(ds.get_default_workflow(i) for i in ids))
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_get_default_workflow_throughput_medium_load():
    """Throughput: medium load (100 requests)."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    ids = list(range(100))
    results = await asyncio.gather(*(ds.get_default_workflow(i) for i in ids))
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_get_default_workflow_throughput_varying_parameters():
    """Throughput: requests with varying parameters."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    # Mix of ids, returnDraftIfExists, and headers
    coros = [
        ds.get_default_workflow(
            i, returnDraftIfExists=(i % 2 == 0), headers={"X-Load": str(i)}
        )
        for i in range(20)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_get_default_workflow_throughput_high_volume():
    """Throughput: high volume (200 requests)."""
    client = MockAsyncClient(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    ids = list(range(200))
    results = await asyncio.gather(*(ds.get_default_workflow(i) for i in ids))
    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  # used to run async functions

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


# Mocks and helpers for HTTPRequest/HTTPResponse and clients
class HTTPRequest:
    def __init__(
        self, method, url, headers=None, path_params=None, query_params=None, body=None
    ):
        self.method = method
        self.url = url
        self.headers = headers or {}
        self.path_params = path_params or {}
        self.query_params = query_params or {}
        self.body = body


class HTTPResponse:
    def __init__(self, response):
        self.response = response

    def json(self):
        # Simulate a JSON response
        return self.response


# Minimal mock for JiraRESTClientViaToken
class MockAsyncClient:
    async def request(self, method, url, **kwargs):
        # Simulate response based on URL and method
        return {"method": method, "url": url, "kwargs": kwargs}


class MockJiraRESTClientViaToken:
    def __init__(self, base_url):
        self.base_url = base_url
        self.client = MockAsyncClient()

    def get_base_url(self):
        return self.base_url

    async def execute(self, request, **kwargs):
        # Return HTTPResponse with a simulated JSON dict
        response = {
            "method": request.method,
            "url": request.url,
            "headers": request.headers,
            "path_params": request.path_params,
            "query_params": request.query_params,
            "body": request.body,
        }
        return HTTPResponse(response)


# Minimal JiraClient wrapper
class JiraClient:
    def __init__(self, client):
        self.client = client

    def get_client(self):
        return self.client


# ----------------------------
# Unit Tests for JiraDataSource.get_default_workflow
# ----------------------------


@pytest.mark.asyncio
async def test_get_default_workflow_basic():
    """Basic test: Ensure correct response for typical input and await behavior."""
    client = JiraClient(MockJiraRESTClientViaToken("https://example.atlassian.net"))
    ds = JiraDataSource(client)
    resp = await ds.get_default_workflow(42)
    data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_with_headers_and_query():
    """Test with custom headers and query param."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    headers = {"X-Test": "value"}
    resp = await ds.get_default_workflow(100, returnDraftIfExists=True, headers=headers)
    data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_returnDraftIfExists_false():
    """Test with returnDraftIfExists set to False."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    resp = await ds.get_default_workflow(5, returnDraftIfExists=False)
    data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_headers_none():
    """Test with headers set to None."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    resp = await ds.get_default_workflow(7, headers=None)
    data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_edge_invalid_client():
    """Edge case: HTTP client is None, should raise ValueError."""

    class BadClient:
        def get_client(self):
            return None

    client = JiraClient(BadClient())
    with pytest.raises(ValueError) as excinfo:
        JiraDataSource(client)


@pytest.mark.asyncio
async def test_get_default_workflow_edge_missing_base_url():
    """Edge case: HTTP client missing get_base_url method."""

    class NoBaseUrlClient:
        pass

    client = JiraClient(NoBaseUrlClient())
    with pytest.raises(ValueError) as excinfo:
        JiraDataSource(client)


@pytest.mark.asyncio
async def test_get_default_workflow_concurrent_calls():
    """Edge case: Multiple concurrent calls should work and be independent."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    tasks = [
        ds.get_default_workflow(i, returnDraftIfExists=(i % 2 == 0)) for i in range(10)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_edge_large_id():
    """Edge case: Large integer ID."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    large_id = 987654321
    resp = await ds.get_default_workflow(large_id)
    data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_edge_empty_headers():
    """Edge case: Empty headers dict."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    resp = await ds.get_default_workflow(123, headers={})
    data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_edge_non_bool_query():
    """Edge case: Non-bool value for returnDraftIfExists (should serialize to string)."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    resp = await ds.get_default_workflow(321, returnDraftIfExists="notabool")
    data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_large_scale_concurrent():
    """Large scale: 50 concurrent requests with varying parameters."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    tasks = [
        ds.get_default_workflow(
            i, returnDraftIfExists=(i % 3 == 0), headers={"X-Req": str(i)}
        )
        for i in range(50)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_large_scale_unique_ids():
    """Large scale: Ensure unique URLs for unique IDs."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    ids = list(range(100, 150))
    tasks = [ds.get_default_workflow(i) for i in ids]
    results = await asyncio.gather(*tasks)
    urls = [resp.json()["url"] for resp in results]


@pytest.mark.asyncio
async def test_get_default_workflow_throughput_small_load():
    """Throughput: Small load, 5 requests."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    tasks = [ds.get_default_workflow(i) for i in range(5)]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_throughput_medium_load():
    """Throughput: Medium load, 50 requests."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    tasks = [ds.get_default_workflow(i, headers={"X-Batch": "med"}) for i in range(50)]
    results = await asyncio.gather(*tasks)
    for resp in results:
        data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_throughput_high_load():
    """Throughput: High load, 200 concurrent requests."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    tasks = [
        ds.get_default_workflow(i, headers={"X-Batch": "high"}) for i in range(200)
    ]
    results = await asyncio.gather(*tasks)
    for resp in results:
        data = resp.json()


@pytest.mark.asyncio
async def test_get_default_workflow_throughput_sustained_pattern():
    """Throughput: Sustained execution pattern with interleaved parameters."""
    client = JiraClient(MockJiraRESTClientViaToken("https://jira.company.com"))
    ds = JiraDataSource(client)
    ids = [10, 20, 30, 40, 50]
    tasks = []
    for i in ids:
        tasks.append(ds.get_default_workflow(i, returnDraftIfExists=True))
        tasks.append(ds.get_default_workflow(i, returnDraftIfExists=False))
    results = await asyncio.gather(*tasks)
    # Check alternating true/false values
    for i in range(0, len(results), 2):
        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_default_workflow-miqa2xm6 and push.