⚡️ Speed up method JiraDataSource.delete_workflow_scheme_draft_issue_type by 8%

[codeflash-ai]

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

⏱️ Runtime : 3.40 milliseconds → 3.15 milliseconds (best of 48 runs)

📝 Explanation and details

The optimized code achieves an 8% runtime speedup through several micro-optimizations that reduce function call overhead and unnecessary object allocations:

Key Optimizations Applied:

1. Conditional header processing: Changed dict(headers or {}) to _as_str_dict(headers) if headers else {}, avoiding unnecessary dict construction and function calls when headers are None/empty.

2. Enhanced _as_str_dict function: Modified to handle None input directly with if d else {}, preventing the need to pass empty dicts through the comprehension.

3. Eliminated redundant header copying in HTTPClient: Replaced unconditional self.headers.copy() with conditional copying only when both self.headers and request.headers exist, reducing memory allocations.

4. Optimized URL formatting: Added conditional check if request.path_params before calling format(**request.path_params), avoiding unnecessary string formatting operations.

Performance Impact Analysis:

The line profiler shows the optimizations primarily benefit the _as_str_dict function, reducing its total execution time from 3.59ms to 2.98ms (17% improvement). The function is called 1,791 times in the original vs 1,243 times in the optimized version, indicating fewer unnecessary calls.

The delete_workflow_scheme_draft_issue_type method shows slight improvements in variable initialization overhead, particularly in the _headers assignment line which now executes more efficiently.

Test Case Benefits:

These optimizations are most effective for:

• High-volume concurrent requests (tests with 50-200 concurrent calls show the cumulative effect)
• Scenarios with empty/None headers (common in many API calls)
• Repeated calls with minimal header customization

While throughput remains constant at 28,704 operations/second, the 8% runtime reduction means lower CPU usage and better resource efficiency for the same workload.

✅ Correctness verification report:

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

🌀 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 HTTPRequest and HTTPResponse
class HTTPRequest:
    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


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


# Minimal stub for JiraClient and its get_client()
class DummyClient:
    def __init__(self, base_url: str):
        self._base_url = base_url
        self.executed_requests = []

    def get_base_url(self):
        return self._base_url

    async def execute(self, request: HTTPRequest):
        # Record the request for inspection
        self.executed_requests.append(request)
        # Simulate a response object
        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:
    def __init__(self, client):
        self.client = client

    def get_client(self):
        return self.client


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


@pytest.mark.asyncio
async def test_basic_delete_workflow_scheme_draft_issue_type_returns_expected_response():
    """Basic test: verify expected response and correct request construction."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    resp = await ds.delete_workflow_scheme_draft_issue_type(123, "bug")


@pytest.mark.asyncio
async def test_basic_with_custom_headers():
    """Basic test: verify custom headers are passed and serialized correctly."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    headers = {"X-Test-Header": "value", "Another-Header": 42}
    resp = await ds.delete_workflow_scheme_draft_issue_type(
        555, "task", headers=headers
    )


@pytest.mark.asyncio
async def test_basic_with_non_ascii_and_special_characters():
    """Basic test: issueType with special/non-ascii characters is handled."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    issue_type = "büg/特殊"
    resp = await ds.delete_workflow_scheme_draft_issue_type(321, issue_type)


@pytest.mark.asyncio
async def test_edge_missing_client_raises_value_error():
    """Edge case: If JiraClient.get_client() returns None, should raise ValueError."""

    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_edge_client_missing_get_base_url_raises_value_error():
    """Edge case: If client lacks get_base_url method, should raise ValueError."""

    class NoBaseUrlClient:
        pass

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


@pytest.mark.asyncio
async def test_edge_delete_workflow_scheme_draft_issue_type_with_none_headers():
    """Edge case: headers=None should be handled gracefully."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    resp = await ds.delete_workflow_scheme_draft_issue_type(1, "story", headers=None)


@pytest.mark.asyncio
async def test_edge_delete_workflow_scheme_draft_issue_type_with_empty_headers():
    """Edge case: headers={} should be handled gracefully."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    resp = await ds.delete_workflow_scheme_draft_issue_type(2, "epic", headers={})


@pytest.mark.asyncio
async def test_edge_concurrent_execution():
    """Edge case: Concurrent execution of several requests."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    ids = [10, 20, 30]
    issue_types = ["bug", "story", "task"]
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(id_, issue_type)
        for id_, issue_type in zip(ids, issue_types)
    ]
    results = await asyncio.gather(*coros)
    # Each result should correspond to its request
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_edge_headers_with_various_types():
    """Edge case: headers with different value types (bool, int, float, None, list)."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    headers = {
        "Bool-Header": True,
        "Int-Header": 123,
        "Float-Header": 1.23,
        "None-Header": None,
        "List-Header": ["a", "b", True, 5],
    }
    resp = await ds.delete_workflow_scheme_draft_issue_type(
        42, "feature", headers=headers
    )


@pytest.mark.asyncio
async def test_large_scale_many_concurrent_requests():
    """Large scale: Run many concurrent requests and verify all responses."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    num_requests = 50  # Large, but not excessive
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(i, f"type{i}")
        for i in range(num_requests)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_large_scale_unique_headers_per_request():
    """Large scale: Each request has unique headers, verify correct assignment."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    num_requests = 25
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(
            i, f"type{i}", headers={"Header": f"value{i}"}
        )
        for i in range(num_requests)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_throughput_small_load():
    """Throughput: Small load, verify all requests complete quickly and correctly."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    coros = [ds.delete_workflow_scheme_draft_issue_type(i, "type") for i in range(5)]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_throughput_medium_load():
    """Throughput: Medium load, verify all requests complete and responses are correct."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(i, f"type{i}") for i in range(20)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_throughput_high_volume():
    """Throughput: High volume, ensure all requests complete and responses are correct."""
    dummy_client = DummyClient("https://jira.example.com")
    ds = JiraDataSource(JiraClient(dummy_client))
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(i, f"bulk{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  # used to run async functions

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

# Mocks and minimal definitions for dependencies


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

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

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


class HTTPRequest:
    """Minimal mock of 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 MockClient:
    """Mock async client for .execute() method."""

    def __init__(self, base_url="http://mockjira.com"):
        self._base_url = base_url
        self.executed_requests = []

    def get_base_url(self):
        return self._base_url

    async def execute(self, req):
        # Capture the request for inspection
        self.executed_requests.append(req)
        # Return a mock HTTPResponse
        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 for testing."""

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

    def get_client(self):
        return self.client


def codeflash_capture(function_name, tmp_dir_path, tests_root, is_fto):
    # Dummy decorator for compatibility
    def decorator(fn):
        return fn

    return decorator


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

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_basic():
    """Test basic async/await behavior and expected output."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    id_val = 123
    issue_type_val = "Bug"
    headers_val = {"X-Test-Header": "testval"}
    # Await the function and verify the HTTPResponse structure
    resp = await ds.delete_workflow_scheme_draft_issue_type(
        id_val, issue_type_val, headers_val
    )


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_no_headers():
    """Test function with no headers argument (should default to empty dict)."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    resp = await ds.delete_workflow_scheme_draft_issue_type(456, "Task")


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_empty_issue_type():
    """Test with empty string for issueType."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    resp = await ds.delete_workflow_scheme_draft_issue_type(789, "")


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_invalid_client_none():
    """Test ValueError raised when JiraDataSource initialized with None client."""

    class DummyJiraClient:
        def get_client(self):
            return None

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


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_invalid_client_missing_base_url():
    """Test ValueError when client does not have get_base_url method."""

    class DummyClient:
        pass

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

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


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_client_none_at_call():
    """Test ValueError raised if _client is None at call time."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    ds._client = None  # forcibly break the client
    with pytest.raises(ValueError, match="HTTP client is not initialized"):
        await ds.delete_workflow_scheme_draft_issue_type(1, "Bug")


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_special_characters():
    """Test issueType and id with special characters."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    id_val = 999
    issue_type_val = "Bug/Feature?Test"
    resp = await ds.delete_workflow_scheme_draft_issue_type(id_val, issue_type_val)


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_concurrent_execution():
    """Test concurrent execution of multiple requests."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    ids = [10, 20, 30]
    issue_types = ["Bug", "Task", "Story"]
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(i, t)
        for i, t in zip(ids, issue_types)
    ]
    results = await asyncio.gather(*coros)
    # Ensure each result corresponds to its input
    for i, t, r in zip(ids, issue_types, results):
        pass


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_large_scale_concurrent():
    """Test function under moderate concurrent load (50 calls)."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    n = 50
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(i, f"IssueType{i}") for i in range(n)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_large_headers():
    """Test with very large headers dict."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    large_headers = {f"Header{i}": f"Value{i}" for i in range(100)}
    resp = await ds.delete_workflow_scheme_draft_issue_type(321, "Epic", large_headers)
    for i in range(100):
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_throughput_small_load():
    """Throughput: test small load of requests (10 calls)."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(i, f"Type{i}") for i in range(10)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_throughput_medium_load():
    """Throughput: test medium load of requests (100 calls)."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(i, f"Type{i}") for i in range(100)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_throughput_varying_headers():
    """Throughput: test with varying headers per request."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(
            i, f"Type{i}", {"Header": f"Value{i}"}
        )
        for i in range(20)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_delete_workflow_scheme_draft_issue_type_throughput_high_volume():
    """Throughput: test high volume (200 concurrent calls, bounded)."""
    mock_client = MockClient()
    jira_client = JiraClient(mock_client)
    ds = JiraDataSource(jira_client)
    coros = [
        ds.delete_workflow_scheme_draft_issue_type(i, f"HighLoadType{i}")
        for i in range(200)
    ]
    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.

To edit these changes git checkout codeflash/optimize-JiraDataSource.delete_workflow_scheme_draft_issue_type-miqe1gui and push.