⚡️ Speed up method JiraDataSource.app_issue_field_value_update_resource_update_issue_fields_put by 22%

[codeflash-ai]

📄 22% (0.22x) speedup for JiraDataSource.app_issue_field_value_update_resource_update_issue_fields_put in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 1.60 milliseconds → 1.32 milliseconds (best of 51 runs)

📝 Explanation and details

The optimized code achieves a 21% runtime improvement (1.60ms → 1.32ms) through several targeted micro-optimizations that reduce overhead in object allocation and unnecessary operations:

Key Optimizations Applied:

1. Eliminated unnecessary URL formatting: The original code called _safe_format_url(rel_path, _path) where _path was always empty. The optimized version directly concatenates self.base_url + rel_path, bypassing the format operation entirely. This saves ~18% of the original function's time (line profiler shows this dropped from 18% to 1.9% of total time).

2. Reduced dictionary allocations: Instead of creating three separate empty dictionaries (_path, _query, _body), the optimized version eliminates _path and _query entirely since they're always empty, and uses a conditional expression for _body creation that only allocates the dictionary when updateValueList is provided.

3. Optimized header handling: Replaced dict(headers or {}) + setdefault() with a more direct conditional copy (headers.copy() if headers else {}) followed by an explicit if check for Content-Type, avoiding the overhead of the setdefault method call.

4. Improved _as_str_dict efficiency: Added a check to avoid calling str() on keys that are already strings (k if isinstance(k, str) else str(k)), reducing unnecessary type conversions.

5. Streamlined HTTPRequest creation: Passes empty dictionaries directly instead of calling _as_str_dict on empty collections, eliminating unnecessary function calls.

Performance Impact:
The line profiler shows the main function's total time reduced from 6.02ms to 4.00ms, with the most significant gains coming from eliminating the _safe_format_url overhead and reducing _as_str_dict calls. The _as_str_dict function time also improved from 1.33ms to 0.98ms.

Test Case Effectiveness:
These optimizations are particularly effective for the high-throughput scenarios tested, including concurrent execution (5-50 parallel calls) and large payloads (100+ field updates). The improvements benefit all usage patterns since they target fundamental operations performed in every function call, making this optimization valuable for any workload that frequently calls this Jira API endpoint.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 241 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	93.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

# --- Minimal stubs for dependencies ---


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

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

    def json(self):
        return self.response_data


class HTTPRequest:
    """Stub for HTTPRequest, holds request data for inspection."""

    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 httpx.AsyncClient, only for testing."""

    def __init__(self):
        self.last_request = None

    async def request(self, method, url, **kwargs):
        # Save request for inspection
        self.last_request = {"method": method, "url": url, "kwargs": kwargs}
        # Mimic a response object
        return {"status_code": 200, "data": kwargs.get("json", {})}


class DummyJiraRESTClientViaUsernamePassword:
    """Stub for JiraRESTClientViaUsernamePassword."""

    def __init__(self, base_url, username, password, token_type="Basic"):
        self.base_url = base_url
        self.client = DummyAsyncClient()

    def get_base_url(self):
        return self.base_url

    async def execute(self, request, **kwargs):
        # Simulate an async HTTP call
        resp_obj = await self.client.request(
            request.method,
            f"{request.url}",
            params=request.query_params,
            headers=request.headers,
            json=request.body,
        )
        return HTTPResponse(resp_obj)


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

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

    def get_client(self):
        return self.client


# --- Unit tests for JiraDataSource.app_issue_field_value_update_resource_update_issue_fields_put ---


@pytest.mark.asyncio
async def test_basic_success_case():
    """Basic: Function returns HTTPResponse with correct data for valid input."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    transfer_id = "ABC123"
    update_list = [{"fieldId": "customfield_1", "value": "foo"}]
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        transfer_id, update_list
    )


@pytest.mark.asyncio
async def test_basic_empty_update_list():
    """Basic: Function works when updateValueList is empty."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    transfer_id = "XYZ789"
    update_list = []
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        transfer_id, update_list
    )


@pytest.mark.asyncio
async def test_basic_headers_override():
    """Basic: Custom headers are merged and override defaults."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    transfer_id = "DEF456"
    custom_headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "X-Test": "yes",
    }
    update_list = [{"fieldId": "customfield_2", "value": "bar"}]
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        transfer_id, update_list, headers=custom_headers
    )


@pytest.mark.asyncio
async def test_basic_no_update_value_list():
    """Basic: updateValueList is None, body should be empty dict."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    transfer_id = "NOUPDATELIST"
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        transfer_id
    )


@pytest.mark.asyncio
async def test_edge_missing_client_raises():
    """Edge: If client is None, should raise ValueError."""

    class DummyNoneClient:
        def get_client(self):
            return None

    jira_client = DummyNoneClient()
    with pytest.raises(ValueError):
        JiraDataSource(jira_client)


@pytest.mark.asyncio
async def test_edge_missing_base_url_method_raises():
    """Edge: If client lacks get_base_url, should raise ValueError."""

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

            return NoBaseUrl()

    jira_client = DummyBadClient()
    with pytest.raises(ValueError):
        JiraDataSource(jira_client)


@pytest.mark.asyncio
async def test_edge_concurrent_execution():
    """Edge: Multiple concurrent calls should each succeed."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    transfer_ids = [f"CONCUR_{i}" for i in range(5)]
    update_lists = [
        [{"fieldId": f"customfield_{i}", "value": f"val_{i}"}] for i in range(5)
    ]
    coros = [
        ds.app_issue_field_value_update_resource_update_issue_fields_put(
            transfer_ids[i], update_lists[i]
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_edge_invalid_atlassian_transfer_id():
    """Edge: Function handles unusual Atlassian_Transfer_Id values."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    transfer_id = ""  # Empty string
    update_list = [{"fieldId": "customfield_3", "value": "baz"}]
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        transfer_id, update_list
    )


@pytest.mark.asyncio
async def test_edge_update_value_list_types():
    """Edge: updateValueList with diverse types."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    transfer_id = "TYPECASE"
    update_list = [
        {"fieldId": "customfield_bool", "value": True},
        {"fieldId": "customfield_int", "value": 42},
        {"fieldId": "customfield_none", "value": None},
        {"fieldId": "customfield_list", "value": [1, 2, 3]},
    ]
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        transfer_id, update_list
    )


@pytest.mark.asyncio
async def test_large_scale_many_fields():
    """Large Scale: updateValueList with many fields."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    transfer_id = "LARGE_SCALE"
    update_list = [
        {"fieldId": f"customfield_{i}", "value": f"value_{i}"} for i in range(100)
    ]
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        transfer_id, update_list
    )


@pytest.mark.asyncio
async def test_large_scale_concurrent_many():
    """Large Scale: Many concurrent calls."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    coros = [
        ds.app_issue_field_value_update_resource_update_issue_fields_put(
            f"CONCURRENT_{i}", [{"fieldId": f"cf_{i}", "value": f"val_{i}"}]
        )
        for i in range(20)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_JiraDataSource_app_issue_field_value_update_resource_update_issue_fields_put_throughput_small_load():
    """Throughput: Small load, 5 sequential calls."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    for i in range(5):
        transfer_id = f"THROUGHPUT_SMALL_{i}"
        update_list = [{"fieldId": f"field_{i}", "value": f"value_{i}"}]
        resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
            transfer_id, update_list
        )


@pytest.mark.asyncio
async def test_JiraDataSource_app_issue_field_value_update_resource_update_issue_fields_put_throughput_medium_load():
    """Throughput: Medium load, 20 concurrent calls."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    coros = [
        ds.app_issue_field_value_update_resource_update_issue_fields_put(
            f"THROUGHPUT_MEDIUM_{i}", [{"fieldId": f"field_{i}", "value": 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_JiraDataSource_app_issue_field_value_update_resource_update_issue_fields_put_throughput_large_load():
    """Throughput: Large load, 50 concurrent calls."""
    jira_client = JiraClient(
        DummyJiraRESTClientViaUsernamePassword("http://jira.test", "user", "pass")
    )
    ds = JiraDataSource(jira_client)
    coros = [
        ds.app_issue_field_value_update_resource_update_issue_fields_put(
            f"THROUGHPUT_LARGE_{i}", [{"fieldId": f"field_{i}", "value": f"value_{i}"}]
        )
        for i in range(50)
    ]
    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 dependencies ----


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

    def __init__(self, status_code=200, json_data=None):
        self.status_code = status_code
        self._json_data = json_data or {}

    def json(self):
        return self._json_data


class DummyHTTPClient:
    """A dummy async HTTP client to simulate .execute()."""

    def __init__(self, base_url):
        self._base_url = base_url
        self.requests = []  # Store requests for inspection
        self.should_raise = False
        self.raise_exc = None
        self.response = DummyHTTPResponse()

    def get_base_url(self):
        return self._base_url

    async def execute(self, req):
        self.requests.append(req)
        if self.should_raise:
            raise self.raise_exc or Exception("Dummy Exception")
        return self.response


class DummyJiraClient:
    """Dummy JiraClient with get_client()."""

    def __init__(self, http_client):
        self._http_client = http_client

    def get_client(self):
        return self._http_client


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, dummy):
        self._dummy = dummy

    def json(self):
        return getattr(self._dummy, "json", lambda: {})()


# ---- TESTS ----

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_basic_success_with_update_value_list():
    """Test normal usage with all parameters set, including updateValueList."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)
    update_list = [{"fieldId": "customfield_10000", "value": "foo"}]
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        Atlassian_Transfer_Id="abc123",
        updateValueList=update_list,
        headers={"X-Test": "yes"},
    )
    # The request should include the correct Atlassian-Transfer-Id header
    last_req = dummy_http_client.requests[-1]


@pytest.mark.asyncio
async def test_basic_success_with_minimal_parameters():
    """Test with only the required Atlassian_Transfer_Id parameter."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        Atlassian_Transfer_Id="xyz789"
    )
    last_req = dummy_http_client.requests[-1]


@pytest.mark.asyncio
async def test_basic_success_with_custom_headers():
    """Test that custom headers are merged correctly."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        Atlassian_Transfer_Id="header-test",
        headers={"Content-Type": "application/custom+json", "X-Extra": "1"},
    )
    last_req = dummy_http_client.requests[-1]


# 2. Edge Test Cases


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

    class BadJiraClient:
        def get_client(self):
            return None

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


@pytest.mark.asyncio
async def test_edge_http_client_missing_get_base_url():
    """Test that ValueError is raised if HTTP 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_edge_execute_raises_exception():
    """Test that exceptions from .execute() are propagated."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_http_client.should_raise = True
    dummy_http_client.raise_exc = RuntimeError("fail execute")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)
    with pytest.raises(RuntimeError, match="fail execute"):
        await ds.app_issue_field_value_update_resource_update_issue_fields_put(
            Atlassian_Transfer_Id="err1"
        )


@pytest.mark.asyncio
async def test_edge_update_value_list_none_and_empty():
    """Test with updateValueList=None and updateValueList=[] (should not crash)."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)
    # None
    resp1 = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        Atlassian_Transfer_Id="none"
    )
    last_req1 = dummy_http_client.requests[-1]
    # Empty list
    resp2 = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        Atlassian_Transfer_Id="empty", updateValueList=[]
    )
    last_req2 = dummy_http_client.requests[-1]


@pytest.mark.asyncio
async def test_edge_concurrent_execution():
    """Test concurrent execution of multiple requests."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)

    async def call(idx):
        return await ds.app_issue_field_value_update_resource_update_issue_fields_put(
            Atlassian_Transfer_Id=f"concurrent-{idx}",
            updateValueList=[{"fieldId": f"f{idx}", "value": idx}],
        )

    results = await asyncio.gather(*[call(i) for i in range(5)])
    # Check that each request was sent with the correct transfer id
    for i in range(5):
        pass


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_large_scale_many_concurrent_requests():
    """Test 50 concurrent requests to check scalability."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)

    async def call(idx):
        return await ds.app_issue_field_value_update_resource_update_issue_fields_put(
            Atlassian_Transfer_Id=f"large-{idx}",
            updateValueList=[{"fieldId": f"f{idx}", "value": idx}],
        )

    tasks = [call(i) for i in range(50)]
    results = await asyncio.gather(*tasks)
    for i in range(50):
        pass


@pytest.mark.asyncio
async def test_large_scale_large_update_value_list():
    """Test with a large updateValueList (500 items)."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)
    update_list = [{"fieldId": f"f{i}", "value": i} for i in range(500)]
    resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
        Atlassian_Transfer_Id="bulk500", updateValueList=update_list
    )
    last_req = dummy_http_client.requests[-1]


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_JiraDataSource_app_issue_field_value_update_resource_update_issue_fields_put_throughput_small_load():
    """Throughput test: small load (10 sequential requests)."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)
    for i in range(10):
        resp = await ds.app_issue_field_value_update_resource_update_issue_fields_put(
            Atlassian_Transfer_Id=f"tp-small-{i}",
            updateValueList=[{"fieldId": f"f{i}", "value": i}],
        )
        last_req = dummy_http_client.requests[-1]


@pytest.mark.asyncio
async def test_JiraDataSource_app_issue_field_value_update_resource_update_issue_fields_put_throughput_medium_concurrent():
    """Throughput test: medium concurrent load (30 concurrent requests)."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)

    async def call(idx):
        return await ds.app_issue_field_value_update_resource_update_issue_fields_put(
            Atlassian_Transfer_Id=f"tp-med-{idx}",
            updateValueList=[{"fieldId": f"f{idx}", "value": idx}],
        )

    results = await asyncio.gather(*[call(i) for i in range(30)])
    for i in range(30):
        pass


@pytest.mark.asyncio
async def test_JiraDataSource_app_issue_field_value_update_resource_update_issue_fields_put_throughput_large_payload():
    """Throughput test: large payload (100 items in updateValueList, 5 concurrent)."""
    dummy_http_client = DummyHTTPClient("https://jira.example.com")
    dummy_jira_client = DummyJiraClient(dummy_http_client)
    ds = JiraDataSource(dummy_jira_client)
    large_update = [{"fieldId": f"f{i}", "value": i} for i in range(100)]

    async def call(idx):
        return await ds.app_issue_field_value_update_resource_update_issue_fields_put(
            Atlassian_Transfer_Id=f"tp-large-{idx}", updateValueList=large_update
        )

    results = await asyncio.gather(*[call(i) for i in range(5)])
    for i in range(5):
        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.app_issue_field_value_update_resource_update_issue_fields_put-miqry4xa and push.