⚡️ Speed up method JiraDataSource.update_security_level by 13%

[codeflash-ai]

📄 13% (0.13x) speedup for JiraDataSource.update_security_level in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 2.28 milliseconds → 2.03 milliseconds (best of 250 runs)

📝 Explanation and details

The optimized code achieves a 12% runtime improvement through several key micro-optimizations that reduce unnecessary object allocations and computations:

Key Optimizations Applied

1. Conditional Dictionary Creation

• Original: Always creates dict(headers or {}) even when headers is None
• Optimized: Uses dict(headers) if headers else {} to avoid unnecessary dict creation when no headers are provided
• This eliminates redundant object allocation in the common case where headers are not passed

2. Eliminated Redundant Variables

• Original: Creates _query: Dict[str, Any] = {} and later calls _as_str_dict(_query)
• Optimized: Directly passes {} to query_params since this endpoint has no query parameters
• Saves one dictionary allocation and one function call per request

3. Removed Redundant Client Check

• Original: Checks if self._client is None: in the method body
• Optimized: Removes this check since the constructor already validates client initialization
• Eliminates an unnecessary conditional check per request

4. Simplified Local Variable Check

• Original: Uses if 'body_additional' in locals() and body_additional:
• Optimized: Uses if body_additional: directly
• Removes the expensive locals() call and dictionary lookup

5. HTTP Client Header Optimization

• Original: Always creates merged headers dictionary: {**self.headers, **request.headers}
• Optimized: Only merges headers when request.headers exists, otherwise uses self.headers directly
• Reduces dictionary creation and unpacking operations

Performance Impact

The line profiler shows the most significant gains come from reducing calls to _as_str_dict (from 1,317 to 878 hits) and eliminating the expensive locals() check. While individual optimizations seem small, they compound effectively since this appears to be in a request processing path that benefits from reduced per-request overhead.

The 12% runtime improvement with 0% throughput change suggests these optimizations primarily reduce CPU cycles per operation rather than changing the fundamental async execution pattern - exactly what you'd expect from eliminating object allocations and redundant checks in a hot path.

✅ Correctness verification report:

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

🌀 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."""
def init(self, data):
self.data = data

class HTTPRequest:
"""Stub for 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

--- Mocks for client and JiraClient ---

class MockHttpClient:
"""A mock HTTP client for testing."""
def init(self, base_url="https://mock-jira.com", execute_result=None, raise_on_execute=None):
self._base_url = base_url
self._execute_result = execute_result
self._raise_on_execute = raise_on_execute
self.requests = []

def get_base_url(self):
    return self._base_url

async def execute(self, req):
    self.requests.append(req)
    if self._raise_on_execute:
        raise self._raise_on_execute
    # Return a stub HTTPResponse with the request data for inspection
    return self._execute_result or 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:
"""A mock JiraClient that returns a mock HTTP client."""
def init(self, client):
self.client = client

def get_client(self):
    return self.client

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

--- TESTS ---

1. Basic Test Cases

@pytest.mark.asyncio
async def test_update_security_level_basic_minimal():
"""Test basic async call with minimal required arguments."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
resp = await ds.update_security_level("123", "456")

@pytest.mark.asyncio
async def test_update_security_level_basic_with_all_fields():
"""Test with all optional fields provided."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
resp = await ds.update_security_level(
"schemeX", "levelY",
description="desc",
name="levelName",
body_additional={"foo": "bar", "baz": 123},
headers={"Authorization": "Bearer token"}
)

@pytest.mark.asyncio
async def test_update_security_level_basic_async_behavior():
"""Test that the function returns a coroutine and works with await."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
codeflash_output = ds.update_security_level("1", "2"); coro = codeflash_output
resp = await coro

2. Edge Test Cases

@pytest.mark.asyncio
async def test_update_security_level_edge_empty_strings():
"""Test with empty string for schemeId and levelId."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
resp = await ds.update_security_level("", "")

@pytest.mark.asyncio
async def test_update_security_level_edge_body_additional_overrides():
"""Test that body_additional can override description/name."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
resp = await ds.update_security_level(
"id1", "id2",
description="desc",
name="n",
body_additional={"description": "override", "name": "override_name"}
)

@pytest.mark.asyncio
async def test_update_security_level_edge_headers_override():
"""Test that headers passed override Content-Type if set."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
resp = await ds.update_security_level(
"id1", "id2",
headers={"Content-Type": "application/xml"}
)

@pytest.mark.asyncio
async def test_update_security_level_edge_client_none_raises():
"""Test that ValueError is raised if client.get_client() returns None."""
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_update_security_level_edge_client_no_get_base_url():
"""Test that ValueError is raised if client lacks get_base_url()."""
class NoBaseUrlClient:
pass
class FakeJiraClient:
def get_client(self):
return NoBaseUrlClient()
with pytest.raises(ValueError, match="does not have get_base_url"):
JiraDataSource(FakeJiraClient())

@pytest.mark.asyncio
async def test_update_security_level_edge_execute_raises():
"""Test that exceptions in client.execute propagate."""
exc = RuntimeError("execute failed")
mock_client = MockHttpClient(raise_on_execute=exc)
ds = JiraDataSource(JiraClient(mock_client))
with pytest.raises(RuntimeError, match="execute failed"):
await ds.update_security_level("a", "b")

@pytest.mark.asyncio
async def test_update_security_level_edge_concurrent_calls():
"""Test concurrent execution of the async function."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
# Run 5 concurrent calls with different parameters
results = await asyncio.gather(
ds.update_security_level("1", "2"),
ds.update_security_level("3", "4", description="d"),
ds.update_security_level("5", "6", name="n"),
ds.update_security_level("7", "8", body_additional={"x": "y"}),
ds.update_security_level("9", "10", headers={"X-Test": "yes"})
)

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_update_security_level_large_scale_concurrent():
"""Test 50 concurrent calls to check async scalability."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
tasks = [
ds.update_security_level(str(i), str(i+1), description=f"desc{i}")
for i in range(50)
]
results = await asyncio.gather(*tasks)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_update_security_level_large_scale_body_fields():
"""Test with various combinations of body_additional and fields."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
tasks = []
for i in range(10):
tasks.append(ds.update_security_level(
f"sch{i}", f"lvl{i}",
description=None if i % 2 == 0 else f"desc{i}",
name=None if i % 3 == 0 else f"name{i}",
body_additional={"extra": i} if i % 4 == 0 else None
))
results = await asyncio.gather(*tasks)
for i, resp in enumerate(results):
if i % 2 == 0:
pass
else:
pass
if i % 3 == 0:
pass
else:
pass
if i % 4 == 0:
pass

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_update_security_level_throughput_small_load():
"""Throughput: 5 quick concurrent calls."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
results = await asyncio.gather(*[
ds.update_security_level(str(i), str(i+1))
for i in range(5)
])

@pytest.mark.asyncio
async def test_update_security_level_throughput_medium_load():
"""Throughput: 20 concurrent calls with varied data."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
results = await asyncio.gather(*[
ds.update_security_level(
f"sch{i}", f"lvl{i}",
description=f"d{i}" if i % 2 else None,
name=f"n{i}" if i % 3 else None,
body_additional={"k": i} if i % 4 == 0 else None
)
for i in range(20)
])
for i, resp in enumerate(results):
if i % 2:
pass
if i % 3:
pass
if i % 4 == 0:
pass

@pytest.mark.asyncio
async def test_update_security_level_throughput_high_volume():
"""Throughput: 100 concurrent calls to check sustained async throughput."""
mock_client = MockHttpClient()
ds = JiraDataSource(JiraClient(mock_client))
results = await asyncio.gather(*[
ds.update_security_level(str(i), str(i+1), body_additional={"index": i})
for i in range(100)
])
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

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

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 = json_data
self.text = text_data or ""
def json(self):
return self._json

class HTTPRequest:
"""Stub for HTTPRequest, only stores params 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 DummyHTTPClient:
"""Dummy async HTTP client for simulating .execute()."""
def init(self, base_url="https://jira.example.com", fail_execute=False):
self._base_url = base_url
self.fail_execute = fail_execute
self.last_request = None
def get_base_url(self):
return self._base_url
async def execute(self, req):
self.last_request = req
if self.fail_execute:
raise RuntimeError("Simulated HTTP error")
# Return a dummy HTTPResponse with the body echoed for inspection
return HTTPResponse(status_code=200, json_data={"echo": req.body, "url": req.url, "headers": req.headers})

class JiraClient:
"""Stub JiraClient, returns the dummy HTTP client."""
def init(self, client):
self.client = client
def get_client(self):
return self.client
from app.sources.external.jira.jira import JiraDataSource

--- TESTS ---

1. BASIC TEST CASES

@pytest.mark.asyncio
async def test_update_security_level_basic_minimal():
"""Test basic call with minimal required arguments."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
resp = await ds.update_security_level("123", "456")

@pytest.mark.asyncio
async def test_update_security_level_basic_full_args():
"""Test call with all arguments provided."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
resp = await ds.update_security_level(
schemeId="abc",
levelId="def",
description="desc",
name="level name",
body_additional={"foo": "bar"},
headers={"X-Test": "yes"}
)
# Should include all fields in the echoed body
body = resp.json()["echo"]
# Headers should merge and include Content-Type and X-Test
headers = resp.json()["headers"]

@pytest.mark.asyncio
async def test_update_security_level_basic_body_additional_merges():
"""Test that body_additional merges with other fields and overwrites if needed."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
resp = await ds.update_security_level(
"x", "y",
description="desc",
body_additional={"description": "override", "extra": 42}
)
body = resp.json()["echo"]

2. EDGE TEST CASES

@pytest.mark.asyncio
async def test_update_security_level_invalid_client_none():
"""Test that ValueError is raised if client is None."""
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_update_security_level_invalid_client_missing_method():
"""Test that ValueError is raised if client lacks get_base_url."""
class NoBaseUrlClient:
pass
class DummyJiraClient:
def get_client(self):
return NoBaseUrlClient()
with pytest.raises(ValueError, match="does not have get_base_url"):
JiraDataSource(DummyJiraClient())

@pytest.mark.asyncio
async def test_update_security_level_headers_are_copied_not_mutated():
"""Test that headers passed in are not mutated by the function."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
headers = {"X-Orig": "orig"}
orig_headers = headers.copy()
await ds.update_security_level("1", "2", headers=headers)

@pytest.mark.asyncio
async def test_update_security_level_body_additional_none_and_empty():
"""Test that body_additional=None and body_additional={} are handled correctly."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
# None
resp1 = await ds.update_security_level("1", "2", body_additional=None)
# Empty dict
resp2 = await ds.update_security_level("1", "2", body_additional={})

@pytest.mark.asyncio
async def test_update_security_level_concurrent_execution():
"""Test concurrent execution of the async function."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
# Run 5 concurrent updates with different ids
results = await asyncio.gather(
*[ds.update_security_level(str(i), str(i+1), description=f"desc{i}") for i in range(5)]
)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_update_security_level_execute_exception():
"""Test that exceptions in the underlying client are propagated."""
client = DummyHTTPClient(fail_execute=True)
ds = JiraDataSource(JiraClient(client))
with pytest.raises(RuntimeError, match="Simulated HTTP error"):
await ds.update_security_level("1", "2")

3. LARGE SCALE TEST CASES

@pytest.mark.asyncio
async def test_update_security_level_large_scale_many_concurrent():
"""Test many concurrent calls (scalability, under 100 calls)."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
N = 50 # Large but not excessive
tasks = [
ds.update_security_level(str(i), str(i+1), name=f"name{i}", description=f"desc{i}")
for i in range(N)
]
results = await asyncio.gather(*tasks)
for i, resp in enumerate(results):
body = resp.json()["echo"]

@pytest.mark.asyncio
async def test_update_security_level_large_scale_unique_urls():
"""Ensure all concurrent calls generate unique URLs as expected."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
N = 20
results = await asyncio.gather(
*[ds.update_security_level(f"scheme{i}", f"level{i}") for i in range(N)]
)
urls = [resp.json()["url"] for resp in results]
for i, url in enumerate(urls):
pass

4. THROUGHPUT TEST CASES

@pytest.mark.asyncio
async def test_update_security_level_throughput_small_load():
"""Throughput: small load of 10 requests."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
N = 10
tasks = [ds.update_security_level(str(i), str(i+1)) for i in range(N)]
results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_update_security_level_throughput_medium_load():
"""Throughput: medium load of 50 requests."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
N = 50
tasks = [ds.update_security_level(str(i), str(i+1), name=f"name{i}") for i in range(N)]
results = await asyncio.gather(*tasks)
# All names should match
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_update_security_level_throughput_high_volume():
"""Throughput: high volume (but bounded) of 100 requests."""
client = DummyHTTPClient()
ds = JiraDataSource(JiraClient(client))
N = 100
# Use different body_additional for each
tasks = [
ds.update_security_level(
str(i), str(i+1),
body_additional={"extra": i}
)
for i in range(N)
]
results = await asyncio.gather(*tasks)
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.

To edit these changes git checkout codeflash/optimize-JiraDataSource.update_security_level-mhrx7bva and push.