⚡️ Speed up method JiraDataSource.delete_security_scheme by 7%

[codeflash-ai]

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

⏱️ Runtime : 2.14 milliseconds → 2.01 milliseconds (best of 250 runs)

📝 Explanation and details

The optimization applies conditional dictionary processing to avoid unnecessary function calls when dictionaries are empty. Specifically, in the delete_security_scheme method:

Key Changes:

• headers=_as_str_dict(_headers) if _headers else {} - Only processes headers if they exist
• query_params=_as_str_dict(_query) if _query else {} - Only processes query params if they exist

Why This Improves Performance:
The _as_str_dict function creates a new dictionary via comprehension ({str(k): _serialize_value(v) for k, v in d.items()}), which has overhead even for empty dictionaries. By adding conditional checks, we avoid:

• Dictionary comprehension overhead for empty dicts
• Function call overhead
• Iterator creation and processing

Performance Impact:
The profiler shows a 21% reduction in _as_str_dict calls (from 1800 to 642 hits), directly correlating with the 6% runtime improvement (2.14ms → 2.01ms). The optimization is most effective when _headers or _query dictionaries are frequently empty, which is common in API calls where optional parameters aren't always provided.

Workload Benefits:
This micro-optimization is particularly valuable for high-volume API scenarios where the method is called repeatedly with minimal headers/query parameters. The test results show consistent improvements across all load patterns (basic, concurrent, and throughput tests), indicating the optimization benefits both single calls and batch operations without affecting correctness or async behavior.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 624 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 HTTPRequest/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 stub for JiraRESTClientViaApiKey ----

class JiraRESTClientViaApiKey:
def init(self, base_url: str, email: str, api_key: str) -> None:
self.base_url = base_url
self._executed_requests = []
def get_base_url(self) -> str:
return self.base_url
async def execute(self, request: HTTPRequest):
# Simulate a successful HTTPResponse with request info for test validation
self._executed_requests.append(request)
# Simulate different responses based on schemeId for edge case testing
if request.path_params.get('schemeId') == "raise_error":
raise RuntimeError("Simulated client error")
# Return a dummy HTTPResponse with the request as the "response"
return HTTPResponse(request)

---- Minimal stub for JiraClient ----

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

---- UNIT TESTS ----

@pytest.mark.asyncio
async def test_delete_security_scheme_basic_success():
"""Basic: Should return HTTPResponse for valid schemeId and headers."""
client = JiraRESTClientViaApiKey("http://test.atlassian.net", "user", "key")
ds = JiraDataSource(JiraClient(client))
resp = await ds.delete_security_scheme("1234", headers={"X-Test": "yes"})

@pytest.mark.asyncio
async def test_delete_security_scheme_basic_no_headers():
"""Basic: Should work when headers is None."""
client = JiraRESTClientViaApiKey("http://test.atlassian.net", "user", "key")
ds = JiraDataSource(JiraClient(client))
resp = await ds.delete_security_scheme("5678")

@pytest.mark.asyncio
async def test_delete_security_scheme_basic_empty_schemeId():
"""Edge: Should handle empty schemeId gracefully."""
client = JiraRESTClientViaApiKey("http://test.atlassian.net", "user", "key")
ds = JiraDataSource(JiraClient(client))
resp = await ds.delete_security_scheme("")

@pytest.mark.asyncio
async def test_delete_security_scheme_basic_special_chars_schemeId():
"""Edge: Should handle special characters in schemeId."""
client = JiraRESTClientViaApiKey("http://test.atlassian.net", "user", "key")
ds = JiraDataSource(JiraClient(client))
special_id = "abc/def?ghi"
resp = await ds.delete_security_scheme(special_id)

@pytest.mark.asyncio
async def test_delete_security_scheme_edge_missing_client():
"""Edge: Should raise ValueError if JiraClient returns None for get_client()."""
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_delete_security_scheme_edge_missing_base_url_method():
"""Edge: Should raise ValueError if client lacks get_base_url()."""
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_delete_security_scheme_concurrent_execution():
"""Edge: Should support concurrent async execution."""
client = JiraRESTClientViaApiKey("http://test.atlassian.net", "user", "key")
ds = JiraDataSource(JiraClient(client))
# Launch several concurrent requests with different schemeIds
ids = ["id1", "id2", "id3", "id4", "id5"]
results = await asyncio.gather(
*(ds.delete_security_scheme(sid) for sid in ids)
)
# Each result should have the correct schemeId in path_params
for idx, r in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_security_scheme_large_scale_concurrent():
"""Large Scale: Test with 50 concurrent requests."""
client = JiraRESTClientViaApiKey("http://test.atlassian.net", "user", "key")
ds = JiraDataSource(JiraClient(client))
ids = [f"id_{i}" for i in range(50)]
results = await asyncio.gather(
*(ds.delete_security_scheme(sid) for sid in ids)
)
for idx, r in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_security_scheme_large_scale_varied_headers():
"""Large Scale: Test with varied headers for each request."""
client = JiraRESTClientViaApiKey("http://test.atlassian.net", "user", "key")
ds = JiraDataSource(JiraClient(client))
ids = [f"id_{i}" for i in range(20)]
headers_list = [{"X-Req": str(i)} for i in range(20)]
results = await asyncio.gather(
*(ds.delete_security_scheme(sid, headers=h) for sid, h in zip(ids, headers_list))
)
for idx, r in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_security_scheme_throughput_small_load():
"""Throughput: Test with small load (10 requests)."""
client = JiraRESTClientViaApiKey("http://test.atlassian.net", "user", "key")
ds = JiraDataSource(JiraClient(client))
ids = [f"small_{i}" for i in range(10)]
results = await asyncio.gather(*(ds.delete_security_scheme(sid) for sid in ids))

@pytest.mark.asyncio
async def test_delete_security_scheme_throughput_medium_load():
"""Throughput: Test with medium load (100 requests)."""
client = JiraRESTClientViaApiKey("http://test.atlassian.net", "user", "key")
ds = JiraDataSource(JiraClient(client))
ids = [f"medium_{i}" for i in range(100)]
results = await asyncio.gather(*(ds.delete_security_scheme(sid) for sid in ids))

@pytest.mark.asyncio
async def test_delete_security_scheme_throughput_large_load():
"""Throughput: Test with large load (200 requests)."""
client = JiraRESTClientViaApiKey("http://test.atlassian.net", "user", "key")
ds = JiraDataSource(JiraClient(client))
ids = [f"large_{i}" for i in range(200)]
results = await asyncio.gather(*(ds.delete_security_scheme(sid) for sid in ids))
# Check a few random ids for correctness
for idx in [0, 50, 199]:
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
from typing import Any, Dict, Optional

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

---- Helper classes for JiraClient and HTTPClient ----

class DummyAsyncClient:
"""Dummy async client to simulate httpx.AsyncClient behavior"""
async def request(self, method, url, **kwargs):
# Simulate a response object
return {
'method': method,
'url': url,
'headers': kwargs.get('headers', {}),
'params': kwargs.get('params', {}),
'body': kwargs.get('json', None) or kwargs.get('data', None) or kwargs.get('content', None),
}

class DummyHTTPClient:
"""Dummy HTTP client mimicking HTTPClient for testing"""
def init(self, base_url="https://example.atlassian.net", should_fail=False):
self.base_url = base_url
self.should_fail = should_fail
self.client = DummyAsyncClient()
def get_base_url(self):
return self.base_url
async def execute(self, request: HTTPRequest, **kwargs):
if self.should_fail:
raise RuntimeError("Simulated failure in execute")
resp = await self.client.request(request.method, request.url, headers=request.headers, params=request.query_params, json=request.body)
return HTTPResponse(resp)

class JiraRESTClientViaApiKey(DummyHTTPClient):
def init(self, base_url: str, email: str, api_key: str, should_fail=False):
super().init(base_url, should_fail=should_fail)
self.email = email
self.api_key = api_key

class JiraClient:
def init(self, client: DummyHTTPClient):
self.client = client
def get_client(self):
return self.client
from app.sources.external.jira.jira import JiraDataSource

---- Unit Tests ----

-------------------- Basic Test Cases --------------------

@pytest.mark.asyncio
async def test_delete_security_scheme_basic_success():
"""Basic: Should return HTTPResponse object with correct method and URL."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey"))
ds = JiraDataSource(client)
scheme_id = "12345"
resp = await ds.delete_security_scheme(scheme_id)

@pytest.mark.asyncio
async def test_delete_security_scheme_basic_with_headers():
"""Basic: Should include custom headers in the request."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey"))
ds = JiraDataSource(client)
scheme_id = "abcde"
custom_headers = {"X-Test-Header": "test-value"}
resp = await ds.delete_security_scheme(scheme_id, headers=custom_headers)

@pytest.mark.asyncio
async def test_delete_security_scheme_basic_empty_headers():
"""Basic: Should handle empty headers gracefully."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey"))
ds = JiraDataSource(client)
scheme_id = "00000"
resp = await ds.delete_security_scheme(scheme_id, headers={})

-------------------- Edge Test Cases --------------------

@pytest.mark.asyncio
async def test_delete_security_scheme_invalid_client_raises():
"""Edge: Should raise ValueError if HTTP client is not initialized."""
class DummyNoneClient:
def get_client(self):
return None
with pytest.raises(ValueError, match="HTTP client is not initialized"):
JiraDataSource(DummyNoneClient())

@pytest.mark.asyncio
async def test_delete_security_scheme_client_missing_base_url_method():
"""Edge: Should raise ValueError if client lacks get_base_url method."""
class DummyBadClient:
def get_client(self):
return object() # no get_base_url
with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
JiraDataSource(DummyBadClient())

@pytest.mark.asyncio
async def test_delete_security_scheme_execute_raises_runtime_error():
"""Edge: Should propagate exceptions from HTTPClient.execute."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey", should_fail=True))
ds = JiraDataSource(client)
with pytest.raises(RuntimeError, match="Simulated failure in execute"):
await ds.delete_security_scheme("fail-id")

@pytest.mark.asyncio
async def test_delete_security_scheme_concurrent_execution():
"""Edge: Should handle multiple concurrent async calls correctly."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey"))
ds = JiraDataSource(client)
ids = ["id1", "id2", "id3", "id4", "id5"]
results = await asyncio.gather(*(ds.delete_security_scheme(sid) for sid in ids))
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_security_scheme_schemeid_special_characters():
"""Edge: Should handle schemeId with special characters."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey"))
ds = JiraDataSource(client)
scheme_id = "id/with/special?chars&=!"
resp = await ds.delete_security_scheme(scheme_id)

-------------------- Large Scale Test Cases --------------------

@pytest.mark.asyncio
async def test_delete_security_scheme_large_scale_concurrent():
"""Large: Should handle 50 concurrent deletions correctly."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey"))
ds = JiraDataSource(client)
ids = [f"scheme-{i}" for i in range(50)]
results = await asyncio.gather(*(ds.delete_security_scheme(sid) for sid in ids))
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_delete_security_scheme_large_scale_unique_headers():
"""Large: Should handle concurrent calls with unique headers."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey"))
ds = JiraDataSource(client)
ids = [f"scheme-{i}" for i in range(20)]
headers_list = [{"X-Req": str(i)} for i in range(20)]
coros = [ds.delete_security_scheme(ids[i], headers=headers_list[i]) for i in range(20)]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
pass

-------------------- Throughput Test Cases --------------------

@pytest.mark.asyncio
async def test_delete_security_scheme_throughput_small_load():
"""Throughput: Test with a small batch of requests."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey"))
ds = JiraDataSource(client)
ids = [f"small-{i}" for i in range(5)]
results = await asyncio.gather(*(ds.delete_security_scheme(sid) for sid in ids))
for resp in results:
pass

@pytest.mark.asyncio
async def test_delete_security_scheme_throughput_medium_load():
"""Throughput: Test with a medium batch of requests."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey"))
ds = JiraDataSource(client)
ids = [f"medium-{i}" for i in range(25)]
results = await asyncio.gather(*(ds.delete_security_scheme(sid) for sid in ids))
for resp in results:
pass

@pytest.mark.asyncio
async def test_delete_security_scheme_throughput_high_volume():
"""Throughput: Test with a high volume of requests (100)."""
client = JiraClient(JiraRESTClientViaApiKey("https://example.atlassian.net", "user@example.com", "apikey"))
ds = JiraDataSource(client)
ids = [f"high-{i}" for i in range(100)]
results = await asyncio.gather(*(ds.delete_security_scheme(sid) for sid in ids))
for resp in 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_security_scheme-mhrvjupn and push.