⚡️ Speed up method JiraDataSource.delete_issue_type by 14%

[codeflash-ai]

📄 14% (0.14x) speedup for JiraDataSource.delete_issue_type in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 1.86 milliseconds → 1.62 milliseconds (best of 33 runs)

📝 Explanation and details

The optimized code achieves a 14% runtime speedup through several targeted micro-optimizations that reduce function call overhead and unnecessary object creation:

Key Optimizations:

1. Inlined _as_str_dict logic: The original version called _serialize_value() for every dictionary value, creating significant function call overhead. The optimized version inlines this serialization logic directly, handling common types (bool, str, int, float) without function calls. This is particularly effective since the profiler shows _as_str_dict consuming significant time across 1,371 calls.

2. Streamlined _safe_format_url: Changed from template.format_map(_SafeDict(params)) to direct template.format(**params), eliminating the overhead of creating a custom dictionary wrapper class. The profiler shows this reduced execution time from 560μs to 306μs.

3. Optimized dictionary construction:

   • _headers now uses conditional construction (dict(headers) if headers else {}) instead of always calling dict(headers or {})
   • _query uses direct conditional assignment instead of creating an empty dict then conditionally populating it
   • Added local variable client = self._client to avoid repeated attribute lookups

4. HTTP client URL formatting optimization: Only applies string formatting when path_params is non-empty, avoiding unnecessary formatting operations for URLs without parameters.

Performance Impact:
The optimizations are most effective for high-frequency API operations with small parameter sets, as evidenced by the test cases. The throughput remains constant at 15,081 ops/sec because the async nature means the bottleneck is likely I/O rather than CPU processing, but the reduced CPU overhead per operation improves overall efficiency and reduces latency by 14%.

These optimizations are particularly valuable for applications making many concurrent JIRA API calls, where the cumulative effect of reduced per-request overhead becomes significant.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio

import pytest
from app.sources.external.jira.jira import JiraDataSource

---- Minimal stub implementations for dependencies ----

class HTTPResponse:
"""Stub for HTTPResponse, mimics a real HTTP response object."""
def init(self, status_code=200, json_data=None, text_data=None):
self.status_code = status_code
self._json_data = json_data or {}
self.text = text_data or ""
def json(self):
return self._json_data

class HTTPRequest:
"""Stub for HTTPRequest, just holds the request data."""
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 DummyHttpClient:
"""Dummy async client for simulating HTTP interactions."""
def init(self, base_url, should_raise=False, delay=0):
self._base_url = base_url
self.should_raise = should_raise
self.delay = delay
self.requests = []
def get_base_url(self):
return self._base_url
async def execute(self, request):
self.requests.append(request)
if self.should_raise:
raise RuntimeError("Simulated client error")
if self.delay > 0:
await asyncio.sleep(self.delay)
# Simulate response based on input
resp_json = {
"deleted": request.path_params.get("id"),
"alternative": request.query_params.get("alternativeIssueTypeId"),
"headers": request.headers
}
return HTTPResponse(status_code=204, json_data=resp_json, text_data="No Content")

class JiraClient:
"""Stub JiraClient that wraps DummyHttpClient."""
def init(self, client):
self.client = client
def get_client(self):
return self.client
from app.sources.external.jira.jira import JiraDataSource

---- Unit Tests ----

1. Basic Test Cases

@pytest.mark.asyncio
async def test_delete_issue_type_basic_success():
"""Test basic async deletion with valid id and no alternative."""
client = DummyHttpClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(client))
resp = await ds.delete_issue_type("123")

@pytest.mark.asyncio
async def test_delete_issue_type_with_alternative():
"""Test deletion with alternativeIssueTypeId provided."""
client = DummyHttpClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(client))
resp = await ds.delete_issue_type("456", alternativeIssueTypeId="789")

@pytest.mark.asyncio
async def test_delete_issue_type_with_headers():
"""Test deletion with custom headers."""
client = DummyHttpClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(client))
custom_headers = {"X-Test-Header": "testval"}
resp = await ds.delete_issue_type("321", headers=custom_headers)

2. Edge Test Cases

@pytest.mark.asyncio

async def test_delete_issue_type_invalid_client_no_get_base_url():
"""Test ValueError raised if client lacks get_base_url."""
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_delete_issue_type_client_execute_raises():
"""Test that exceptions from client.execute are propagated."""
client = DummyHttpClient("https://jira.example.com", should_raise=True)
ds = JiraDataSource(JiraClient(client))
with pytest.raises(RuntimeError, match="Simulated client error"):
await ds.delete_issue_type("failid")

@pytest.mark.asyncio
async def test_delete_issue_type_empty_id():
"""Test deletion with empty string id."""
client = DummyHttpClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(client))
resp = await ds.delete_issue_type("")

@pytest.mark.asyncio
async def test_delete_issue_type_concurrent_calls():
"""Test concurrent execution of multiple deletions."""
client = DummyHttpClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(client))
ids = ["a", "b", "c", "d"]
coros = [ds.delete_issue_type(i) for i in ids]
results = await asyncio.gather(*coros)
for idx, resp in enumerate(results):
pass

3. Large Scale Test Cases

@pytest.mark.asyncio

async def test_delete_issue_type_throughput_small_load():
"""Throughput: Small load (5 concurrent calls)."""
client = DummyHttpClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(client))
ids = [f"small-{i}" for i in range(5)]
coros = [ds.delete_issue_type(i) for i in ids]
results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_delete_issue_type_throughput_medium_load():
"""Throughput: Medium load (20 concurrent calls)."""
client = DummyHttpClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(client))
ids = [f"medium-{i}" for i in range(20)]
coros = [ds.delete_issue_type(i, alternativeIssueTypeId=f"alt-{i}") for i in ids]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_issue_type_throughput_high_load():
"""Throughput: High load (100 concurrent calls, bounded for speed)."""
client = DummyHttpClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(client))
ids = [f"high-{i}" for i in range(100)]
coros = [ds.delete_issue_type(i) for i in ids]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_issue_type_throughput_sustained_pattern():
"""Throughput: Sustained execution pattern (10 sequential batches of 10)."""
client = DummyHttpClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(client))
for batch in range(10):
ids = [f"sustain-{batch}-{i}" for i in range(10)]
coros = [ds.delete_issue_type(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
from typing import Any, Dict, Optional

import pytest
from app.sources.external.jira.jira import JiraDataSource

--- Minimal stubs for required classes and helpers ---

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: Any):
self.response = response

def json(self):
    return getattr(self.response, "json_data", self.response)

class DummyAsyncClient:
"""A dummy async client to simulate HTTP requests."""
def init(self, base_url: str):
self._base_url = base_url
self.last_request = None
self.should_raise = None
self.response_data = None

def get_base_url(self):
    return self._base_url

async def execute(self, req: HTTPRequest):
    self.last_request = req
    if self.should_raise:
        raise self.should_raise
    # Simulate an HTTPResponse with dummy data
    return HTTPResponse(self.response_data or {
        "status": "deleted",
        "id": req.path_params.get("id"),
        "alternativeIssueTypeId": req.query_params.get("alternativeIssueTypeId")
    })

class JiraRESTClientViaApiKey(DummyAsyncClient):
pass

class JiraClient:
def init(self, client: DummyAsyncClient):
self.client = client

def get_client(self):
    return self.client

from app.sources.external.jira.jira import JiraDataSource

--- Unit Tests ---

1. Basic Test Cases

@pytest.mark.asyncio
async def test_delete_issue_type_basic_success():
"""Test basic deletion with valid id and no alternativeIssueTypeId."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
result = await ds.delete_issue_type("12345")

@pytest.mark.asyncio
async def test_delete_issue_type_with_alternative_issue_type_id():
"""Test deletion with both id and alternativeIssueTypeId."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
alt_id = "67890"
result = await ds.delete_issue_type("12345", alternativeIssueTypeId=alt_id)

@pytest.mark.asyncio
async def test_delete_issue_type_with_custom_headers():
"""Test passing custom headers to the request."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
headers = {"X-Custom-Header": "value"}
result = await ds.delete_issue_type("abcde", headers=headers)

2. Edge Test Cases

@pytest.mark.asyncio

async def test_delete_issue_type_invalid_client_missing_base_url():
"""Test error raised when client does not have get_base_url method."""
class BadClient:
pass
class BadJiraClient:
def get_client(self):
return BadClient()
with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
JiraDataSource(BadJiraClient())

@pytest.mark.asyncio
async def test_delete_issue_type_execute_raises_exception():
"""Test that exceptions from execute are propagated."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
dummy_client.should_raise = RuntimeError("network error")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
with pytest.raises(RuntimeError, match="network error"):
await ds.delete_issue_type("12345")

@pytest.mark.asyncio
async def test_delete_issue_type_empty_id():
"""Test deletion with empty string id."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
result = await ds.delete_issue_type("")

@pytest.mark.asyncio
async def test_delete_issue_type_concurrent_execution():
"""Test concurrent deletion requests with different ids."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
ids = ["id1", "id2", "id3"]
results = await asyncio.gather(*(ds.delete_issue_type(i) for i in ids))
for i, res in zip(ids, results):
pass

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_delete_issue_type_large_scale_concurrent():
"""Test large scale concurrent deletions (up to 50)."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
ids = [f"id_{i}" for i in range(50)]
results = await asyncio.gather(*(ds.delete_issue_type(i) for i in ids))
for i, res in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_issue_type_large_scale_with_alternative_ids():
"""Test large scale concurrent deletions with alternativeIssueTypeId."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
ids = [f"id_{i}" for i in range(30)]
alt_ids = [f"alt_{i}" for i in range(30)]
coros = [ds.delete_issue_type(id_, alternativeIssueTypeId=alt_id)
for id_, alt_id in zip(ids, alt_ids)]
results = await asyncio.gather(*coros)
for i, res in enumerate(results):
pass

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_delete_issue_type_throughput_small_load():
"""Throughput test: small load (5 concurrent deletions)."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
ids = [f"small_{i}" for i in range(5)]
results = await asyncio.gather(*(ds.delete_issue_type(i) for i in ids))
for i, res in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_issue_type_throughput_medium_load():
"""Throughput test: medium load (20 concurrent deletions)."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
ids = [f"medium_{i}" for i in range(20)]
results = await asyncio.gather(*(ds.delete_issue_type(i) for i in ids))
for i, res in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_issue_type_throughput_high_load():
"""Throughput test: high load (100 concurrent deletions)."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
ids = [f"high_{i}" for i in range(100)]
results = await asyncio.gather(*(ds.delete_issue_type(i) for i in ids))
for i, res in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_issue_type_throughput_mixed_parameters():
"""Throughput test: mixed parameters (id, alternativeIssueTypeId, headers)."""
dummy_client = JiraRESTClientViaApiKey("https://jira.example.com")
jira_client = JiraClient(dummy_client)
ds = JiraDataSource(jira_client)
coros = []
for i in range(10):
id_ = f"mix_{i}"
alt_id = f"alt_{i}" if i % 2 == 0 else None
headers = {"X-Test": f"test_{i}"} if i % 3 == 0 else None
coros.append(ds.delete_issue_type(id_, alternativeIssueTypeId=alt_id, headers=headers))
results = await asyncio.gather(*coros)
for i, res in enumerate(results):
if i % 2 == 0:
pass
else:
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_issue_type-mhs06ovw and push.