⚡️ Speed up method JiraDataSource.get_draft_default_workflow by 7%

[codeflash-ai]

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

⏱️ Runtime : 3.00 milliseconds → 2.80 milliseconds (best of 50 runs)

📝 Explanation and details

The optimized code achieves a 7% runtime improvement through two key micro-optimizations in hot path functions:

1. Short-circuit optimization in _safe_format_url:

• Added if '{' not in template: return template check before expensive formatting operations
• This avoids the costly template.format_map(_SafeDict(params)) call when no formatting is needed
• The profiler shows this function takes 20% of total time, making this optimization impactful

2. Early return optimization in _as_str_dict:

• Added if not d: return {} check to skip dictionary comprehension for empty dictionaries
• This function is called 3 times per request (for headers, path_params, query_params) and shows up as 61.5% of total time in profiling
• The profiler results show this optimization reduced processing for 1127 out of 1728 calls (65% hit rate)

Performance Impact:

• Runtime improved from 3.00ms to 2.80ms (7% speedup)
• Throughput remains stable at 28,800 operations/second due to the async nature and I/O bound operations
• The optimizations are most effective for requests with empty headers/query parameters and static URL paths (common in API clients)

Test Case Performance:
The optimizations show consistent benefits across all test scenarios, particularly for:

• High-volume concurrent requests (100+ operations)
• Requests with empty headers dictionaries
• Static API endpoints with no URL templating

These micro-optimizations target the most frequently called utility functions without changing any interfaces or behavior, making them safe for production deployment.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 603 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 required classes and functions ---


class HTTPResponse:
    """Stub for HTTPResponse, mimics a real HTTP response."""

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


class HTTPRequest:
    """Stub for HTTPRequest, mimics a real HTTP request."""

    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 DummyAsyncClient:
    """Stub for an async HTTP client."""

    async def request(self, method, url, **kwargs):
        # Simulate a successful HTTP response
        return {"method": method, "url": url, "kwargs": kwargs}


class DummyHTTPClient:
    """Stub for HTTPClient with async execute method."""

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

    def get_base_url(self):
        return self.base_url

    async def execute(self, request, **kwargs):
        # Simulate returning an HTTPResponse with request info
        return HTTPResponse(
            {
                "method": request.method,
                "url": request.url,
                "headers": request.headers,
                "path_params": request.path_params,
                "query_params": request.query_params,
                "body": request.body,
            }
        )


class JiraClient:
    """Stub for JiraClient builder."""

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

    def get_client(self):
        return self.client


def codeflash_capture(**kwargs):
    # No-op decorator for testing
    def decorator(fn):
        return fn

    return decorator


# --- Unit Tests ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_get_draft_default_workflow_basic():
    """Basic: Test normal usage with valid id and no headers."""
    client = JiraClient(DummyHTTPClient("https://example.atlassian.net"))
    ds = JiraDataSource(client)
    response = await ds.get_draft_default_workflow(123)


@pytest.mark.asyncio
async def test_get_draft_default_workflow_with_headers():
    """Basic: Test usage with custom headers."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    headers = {"X-Test": "value", "Authorization": "Bearer token"}
    response = await ds.get_draft_default_workflow(42, headers=headers)


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_get_draft_default_workflow_id_zero():
    """Edge: Test id=0 (minimum integer value)."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    response = await ds.get_draft_default_workflow(0)


@pytest.mark.asyncio
async def test_get_draft_default_workflow_negative_id():
    """Edge: Test negative id."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    response = await ds.get_draft_default_workflow(-99)


@pytest.mark.asyncio
async def test_get_draft_default_workflow_empty_headers():
    """Edge: Test with empty headers dict."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    response = await ds.get_draft_default_workflow(1, headers={})


@pytest.mark.asyncio
async def test_get_draft_default_workflow_none_client_raises():
    """Edge: Test ValueError 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_get_draft_default_workflow_missing_get_base_url_raises():
    """Edge: Test ValueError raised if client lacks get_base_url."""

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

            return NoBaseUrl()

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


@pytest.mark.asyncio
async def test_get_draft_default_workflow_concurrent():
    """Edge: Test concurrent execution with different ids."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    ids = [10, 20, 30, 40]
    results = await asyncio.gather(*(ds.get_draft_default_workflow(i) for i in ids))
    for idx, response in zip(ids, results):
        pass


@pytest.mark.asyncio
async def test_get_draft_default_workflow_headers_types():
    """Edge: Test headers with non-string values."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    headers = {"X-Int": 123, "X-Bool": True, "X-None": None}
    response = await ds.get_draft_default_workflow(5, headers=headers)


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_get_draft_default_workflow_many_concurrent():
    """Large Scale: Test many concurrent requests."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    ids = list(range(50))  # 50 concurrent calls
    responses = await asyncio.gather(*(ds.get_draft_default_workflow(i) for i in ids))
    for i, resp in enumerate(responses):
        pass


@pytest.mark.asyncio
async def test_get_draft_default_workflow_large_id():
    """Large Scale: Test with a very large id value."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    large_id = 9876543210
    response = await ds.get_draft_default_workflow(large_id)


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_get_draft_default_workflow_throughput_small_load():
    """Throughput: Test small load of 10 requests."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    ids = list(range(10))
    responses = await asyncio.gather(*(ds.get_draft_default_workflow(i) for i in ids))
    for i, resp in enumerate(responses):
        pass


@pytest.mark.asyncio
async def test_get_draft_default_workflow_throughput_medium_load():
    """Throughput: Test medium load of 100 requests."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    ids = list(range(100))
    responses = await asyncio.gather(*(ds.get_draft_default_workflow(i) for i in ids))
    for i, resp in enumerate(responses):
        pass


@pytest.mark.asyncio
async def test_get_draft_default_workflow_throughput_headers_load():
    """Throughput: Test throughput with custom headers on each request."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    ids = list(range(20))
    responses = await asyncio.gather(
        *(ds.get_draft_default_workflow(i, headers={"X-Req": f"id{i}"}) for i in ids)
    )
    for i, resp in enumerate(responses):
        pass


@pytest.mark.asyncio
async def test_get_draft_default_workflow_throughput_high_volume():
    """Throughput: Test high volume (but <1000) requests for scalability."""
    client = JiraClient(DummyHTTPClient("https://jira.test"))
    ds = JiraDataSource(client)
    ids = list(range(200))
    responses = await asyncio.gather(*(ds.get_draft_default_workflow(i) for i in ids))
    for i, resp in enumerate(responses):
        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 minimal stubs for dependencies
class HTTPResponse:
    """Mock HTTPResponse for testing."""

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


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


class MockHTTPClient:
    """Mock HTTP client to simulate async HTTP requests."""

    def __init__(self, base_url="https://jira.example.com"):
        self._base_url = base_url
        self.executed_requests = []

    def get_base_url(self):
        """Return the base URL."""
        return self._base_url

    async def execute(self, request):
        """Simulate async HTTP request execution."""
        self.executed_requests.append(request)
        # Simulate a response based on the request
        return HTTPResponse(
            {
                "method": request.method,
                "url": request.url,
                "headers": request.headers,
                "path_params": request.path_params,
                "query_params": request.query_params,
                "body": request.body,
            }
        )


class JiraClient:
    """Mock JiraClient for testing."""

    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_draft_default_workflow_basic_returns_expected_response():
    """Test basic async/await behavior and expected response structure."""
    mock_client = MockHTTPClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    response = await ds.get_draft_default_workflow(123)


@pytest.mark.asyncio
async def test_get_draft_default_workflow_with_custom_headers():
    """Test passing custom headers and their serialization."""
    mock_client = MockHTTPClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    headers = {"X-Test-Header": "value", "Another": 42}
    response = await ds.get_draft_default_workflow(456, headers=headers)


@pytest.mark.asyncio
async def test_get_draft_default_workflow_with_zero_id():
    """Test edge case with id=0."""
    mock_client = MockHTTPClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    response = await ds.get_draft_default_workflow(0)


# 2. Edge Test Cases


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

    class DummyClient:
        def get_client(self):
            return None

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


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

    class BadClient:
        pass

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


@pytest.mark.asyncio
async def test_get_draft_default_workflow_concurrent_requests():
    """Test concurrent execution of multiple async requests."""
    mock_client = MockHTTPClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    ids = [101, 102, 103, 104, 105]
    # Run requests concurrently
    responses = await asyncio.gather(*(ds.get_draft_default_workflow(i) for i in ids))
    for i, resp in zip(ids, responses):
        pass


@pytest.mark.asyncio
async def test_get_draft_default_workflow_headers_serialization_edge():
    """Test serialization of various header value types."""
    mock_client = MockHTTPClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    headers = {
        "BoolTrue": True,
        "BoolFalse": False,
        "NoneVal": None,
        "ListVal": [1, True, "abc"],
        "SetVal": set([2, False]),
    }
    response = await ds.get_draft_default_workflow(999, headers=headers)
    # SetVal is unordered, but contains '2' and 'false'
    set_val = response.data["headers"]["SetVal"].split(",")


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_get_draft_default_workflow_many_concurrent_requests():
    """Test scalability with many concurrent async requests."""
    mock_client = MockHTTPClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    ids = list(range(10, 60))  # 50 concurrent requests
    responses = await asyncio.gather(*(ds.get_draft_default_workflow(i) for i in ids))
    for i, resp in zip(ids, responses):
        pass


@pytest.mark.asyncio
async def test_get_draft_default_workflow_large_id_and_headers():
    """Test with large id value and large headers dict."""
    mock_client = MockHTTPClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    large_id = 999999999
    large_headers = {f"Header{i}": i for i in range(100)}
    response = await ds.get_draft_default_workflow(large_id, headers=large_headers)
    # All headers should be present and stringified
    for k, v in large_headers.items():
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_get_draft_default_workflow_throughput_small_load():
    """Throughput test: small load, 5 concurrent requests."""
    mock_client = MockHTTPClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    ids = [1, 2, 3, 4, 5]
    responses = await asyncio.gather(*(ds.get_draft_default_workflow(i) for i in ids))
    for i, resp in zip(ids, responses):
        pass


@pytest.mark.asyncio
async def test_get_draft_default_workflow_throughput_medium_load():
    """Throughput test: medium load, 20 concurrent requests."""
    mock_client = MockHTTPClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    ids = list(range(100, 120))
    responses = await asyncio.gather(*(ds.get_draft_default_workflow(i) for i in ids))
    for i, resp in zip(ids, responses):
        pass


@pytest.mark.asyncio
async def test_get_draft_default_workflow_throughput_high_volume():
    """Throughput test: high volume, 100 concurrent requests."""
    mock_client = MockHTTPClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    ids = list(range(1000, 1100))  # 100 requests
    responses = await asyncio.gather(*(ds.get_draft_default_workflow(i) for i in ids))
    for i, resp in zip(ids, responses):
        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_draft_default_workflow-miqcyzl6 and push.