⚡️ Speed up method JiraDataSource.delete_workflow_scheme by 7%

[codeflash-ai]

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

⏱️ Runtime : 3.04 milliseconds → 2.85 milliseconds (best of 51 runs)

📝 Explanation and details

The optimization achieves a 6% runtime improvement through a targeted micro-optimization in the _as_str_dict helper function. The key change adds an early return for empty dictionaries:

def _as_str_dict(d: Dict[str, Any]) -> Dict[str, str]:
    if not d:  # Early return for empty dicts
        return {}
    return {str(k): _serialize_value(v) for (k, v) in d.items()}

Why this optimization works:

• Reduced dictionary comprehension overhead: When dictionaries are empty (common for _query and sometimes _headers), the optimization skips the expensive dictionary comprehension that calls _serialize_value() for each item
• Memory allocation efficiency: Returns a pre-allocated empty dict instead of creating one through comprehension
• Function call elimination: Avoids multiple str() and _serialize_value() calls when unnecessary

Performance impact from line profiler data:

• _as_str_dict function calls reduced from 2.71ms total time to 2.42ms (10% improvement in this helper)
• The function is called 3 times per delete_workflow_scheme invocation (for headers, path_params, query_params)
• Empty dictionary handling is particularly common for _query parameters in this Jira API endpoint

Test case optimization benefits:
The improvement is most significant for test cases with minimal or no headers/query parameters, which represents typical API usage patterns. Tests with large headers or concurrent requests still benefit proportionally from the reduced per-call overhead.

While throughput remains constant at ~31,161 operations/second, the 6% runtime reduction directly translates to lower latency per operation, making this optimization valuable for high-frequency API calls where every microsecond counts.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 637 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


# Mocks for HTTPRequest and HTTPResponse
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


# Minimal JiraRESTClientViaUsernamePassword mock
class JiraRESTClientViaUsernamePassword:
    def __init__(
        self,
        base_url: str,
        username: str = "",
        password: str = "",
        token_type: str = "Basic",
    ) -> None:
        self.base_url = base_url

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

    async def execute(self, request: HTTPRequest):
        # Simulate a response object
        # For test purposes, echo back the request details
        return HTTPResponse(
            {
                "method": request.method,
                "url": request.url,
                "headers": request.headers,
                "path_params": request.path_params,
                "query_params": request.query_params,
                "body": request.body,
            }
        )


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

    def get_client(self):
        return self.client


# =========================
# UNIT TESTS BEGIN HERE
# =========================

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_delete_workflow_scheme_basic():
    """Test basic async/await behavior and correct request construction."""
    client = JiraRESTClientViaUsernamePassword(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    resp = await ds.delete_workflow_scheme(123)


@pytest.mark.asyncio
async def test_delete_workflow_scheme_with_headers():
    """Test passing custom headers."""
    client = JiraRESTClientViaUsernamePassword(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    headers = {"Authorization": "Bearer token", "X-Custom": "abc"}
    resp = await ds.delete_workflow_scheme(42, headers=headers)


@pytest.mark.asyncio
async def test_delete_workflow_scheme_with_zero_id():
    """Test passing id=0 (edge case for id values)."""
    client = JiraRESTClientViaUsernamePassword(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    resp = await ds.delete_workflow_scheme(0)


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_delete_workflow_scheme_invalid_client_none():
    """Test error when client is None."""

    class DummyClient:
        def get_client(self):
            return None

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


@pytest.mark.asyncio
async def test_delete_workflow_scheme_invalid_client_no_base_url():
    """Test error when client lacks get_base_url method."""

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

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


@pytest.mark.asyncio
async def test_delete_workflow_scheme_path_formatting_edge():
    """Test path formatting with unusual id values."""
    client = JiraRESTClientViaUsernamePassword(base_url="https://jira.example.com/")
    ds = JiraDataSource(JiraClient(client))
    # Negative id
    resp = await ds.delete_workflow_scheme(-999)
    # Large id
    resp2 = await ds.delete_workflow_scheme(999999999)


@pytest.mark.asyncio
async def test_delete_workflow_scheme_concurrent_execution():
    """Test concurrent execution of multiple delete_workflow_scheme requests."""
    client = JiraRESTClientViaUsernamePassword(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    ids = [1, 2, 3, 4, 5]
    coros = [ds.delete_workflow_scheme(i) for i in ids]
    results = await asyncio.gather(*coros)
    # Assert all results are correct and unique
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_exception_in_execute():
    """Test that exceptions in .execute are propagated."""

    class FailingClient(JiraRESTClientViaUsernamePassword):
        async def execute(self, request):
            raise RuntimeError("Simulated failure")

    ds = JiraDataSource(JiraClient(FailingClient(base_url="https://jira.example.com")))
    with pytest.raises(RuntimeError, match="Simulated failure"):
        await ds.delete_workflow_scheme(123)


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_delete_workflow_scheme_large_scale_concurrent():
    """Test scalability with 50 concurrent requests."""
    client = JiraRESTClientViaUsernamePassword(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    ids = list(range(100, 150))  # 50 requests
    coros = [ds.delete_workflow_scheme(i) for i in ids]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_large_headers():
    """Test passing a large number of headers."""
    client = JiraRESTClientViaUsernamePassword(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    headers = {f"X-Header-{i}": f"value-{i}" for i in range(100)}
    resp = await ds.delete_workflow_scheme(555, headers=headers)
    for i in range(100):
        pass


# 4. Throughput Test Cases


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


@pytest.mark.asyncio
async def test_delete_workflow_scheme_throughput_medium_load():
    """Throughput test: medium load (100 requests)."""
    client = JiraRESTClientViaUsernamePassword(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    ids = list(range(1000, 1100))
    coros = [ds.delete_workflow_scheme(i) for i in ids]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_throughput_high_volume():
    """Throughput test: high volume (250 requests, below 1000)."""
    client = JiraRESTClientViaUsernamePassword(base_url="https://jira.example.com")
    ds = JiraDataSource(JiraClient(client))
    ids = list(range(5000, 5250))
    coros = [ds.delete_workflow_scheme(i) for i in ids]
    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  # used to run async functions
# The JiraDataSource class under test (copied EXACTLY as provided)

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


# Minimal stubs for HTTPRequest and HTTPResponse, as used by the function
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

    def __eq__(self, other):
        return isinstance(other, HTTPResponse) and self.response == other.response


# Minimal stub for JiraRESTClientViaUsernamePassword
class JiraRESTClientViaUsernamePassword:
    def __init__(self, base_url, username, password, token_type="Basic"):
        self.base_url = base_url

    def get_base_url(self):
        return self.base_url

    async def execute(self, req):
        # Simulate a response object for testing
        # For edge cases, allow the request to specify a special header to trigger exceptions
        if req.headers.get("X-Trigger-Error") == "true":
            raise RuntimeError("Simulated error")
        # Return a mock HTTPResponse with the 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,
            }
        )


# Minimal stub for JiraClient
class JiraClient:
    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_delete_workflow_scheme_basic_success():
    """Test basic successful deletion with valid id and no headers."""
    client = JiraRESTClientViaUsernamePassword(
        "https://jira.example.com", "user", "pass"
    )
    ds = JiraDataSource(JiraClient(client))
    id = 42
    resp = await ds.delete_workflow_scheme(id)


@pytest.mark.asyncio
async def test_delete_workflow_scheme_with_headers():
    """Test deletion with custom headers."""
    client = JiraRESTClientViaUsernamePassword(
        "https://jira.example.com", "user", "pass"
    )
    ds = JiraDataSource(JiraClient(client))
    headers = {"Authorization": "Bearer token", "X-Custom": "value"}
    resp = await ds.delete_workflow_scheme(123, headers)


@pytest.mark.asyncio
async def test_delete_workflow_scheme_id_as_zero():
    """Test deletion with id=0 (edge of valid int)."""
    client = JiraRESTClientViaUsernamePassword(
        "https://jira.example.com", "user", "pass"
    )
    ds = JiraDataSource(JiraClient(client))
    resp = await ds.delete_workflow_scheme(0)


# 2. Edge Test Cases


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

    class DummyClient:
        def get_client(self):
            return None

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


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

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

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


@pytest.mark.asyncio
async def test_delete_workflow_scheme_execute_raises_runtime_error():
    """Test that errors in execute are propagated."""

    class ErrorClient(JiraRESTClientViaUsernamePassword):
        async def execute(self, req):
            raise RuntimeError("Simulated error")

    ds = JiraDataSource(
        JiraClient(ErrorClient("https://jira.example.com", "user", "pass"))
    )
    with pytest.raises(RuntimeError, match="Simulated error"):
        await ds.delete_workflow_scheme(1)


@pytest.mark.asyncio
async def test_delete_workflow_scheme_concurrent_execution():
    """Test concurrent execution of multiple deletions."""
    client = JiraRESTClientViaUsernamePassword(
        "https://jira.example.com", "user", "pass"
    )
    ds = JiraDataSource(JiraClient(client))
    ids = [101, 102, 103]
    results = await asyncio.gather(*(ds.delete_workflow_scheme(i) for i in ids))
    # Validate each response matches its id
    for i, resp in zip(ids, results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_special_headers_trigger_error():
    """Test that special header triggers error in execute."""

    class ErrorTriggerClient(JiraRESTClientViaUsernamePassword):
        async def execute(self, req):
            if req.headers.get("X-Trigger-Error") == "true":
                raise RuntimeError("Triggered error")
            return await super().execute(req)

    ds = JiraDataSource(
        JiraClient(ErrorTriggerClient("https://jira.example.com", "user", "pass"))
    )
    with pytest.raises(RuntimeError, match="Triggered error"):
        await ds.delete_workflow_scheme(999, headers={"X-Trigger-Error": "true"})


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_delete_workflow_scheme_many_concurrent():
    """Test large scale concurrent execution with 50 requests."""
    client = JiraRESTClientViaUsernamePassword(
        "https://jira.example.com", "user", "pass"
    )
    ds = JiraDataSource(JiraClient(client))
    ids = list(range(200, 250))
    results = await asyncio.gather(*(ds.delete_workflow_scheme(i) for i in ids))
    for i, resp in zip(ids, results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_large_id():
    """Test with a very large id value."""
    client = JiraRESTClientViaUsernamePassword(
        "https://jira.example.com", "user", "pass"
    )
    ds = JiraDataSource(JiraClient(client))
    large_id = 2**31 - 1
    resp = await ds.delete_workflow_scheme(large_id)


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_delete_workflow_scheme_throughput_small_load():
    """Throughput test: small load of 5 requests."""
    client = JiraRESTClientViaUsernamePassword(
        "https://jira.example.com", "user", "pass"
    )
    ds = JiraDataSource(JiraClient(client))
    ids = [10, 11, 12, 13, 14]
    results = await asyncio.gather(*(ds.delete_workflow_scheme(i) for i in ids))
    for i, resp in zip(ids, results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_throughput_medium_load():
    """Throughput test: medium load of 25 requests."""
    client = JiraRESTClientViaUsernamePassword(
        "https://jira.example.com", "user", "pass"
    )
    ds = JiraDataSource(JiraClient(client))
    ids = list(range(300, 325))
    results = await asyncio.gather(*(ds.delete_workflow_scheme(i) for i in ids))
    for i, resp in zip(ids, results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_throughput_large_load():
    """Throughput test: large load of 100 requests."""
    client = JiraRESTClientViaUsernamePassword(
        "https://jira.example.com", "user", "pass"
    )
    ds = JiraDataSource(JiraClient(client))
    ids = list(range(400, 500))
    results = await asyncio.gather(*(ds.delete_workflow_scheme(i) for i in ids))
    for i, resp in zip(ids, results):
        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.delete_workflow_scheme-miq6gdlp and push.