⚡️ Speed up method JiraDataSource.set_default_levels by 6%

[codeflash-ai]

📄 6% (0.06x) speedup for JiraDataSource.set_default_levels in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 2.37 milliseconds → 2.23 milliseconds (best of 250 runs)

📝 Explanation and details

The optimized code achieves a 6% runtime improvement through several targeted micro-optimizations that reduce unnecessary object allocations and operations:

Key Optimizations:

1. Conditional header initialization in set_default_levels:

   • Original: dict(headers or {}) always creates a new dict
   • Optimized: headers if headers is not None else {} avoids dict creation when headers is None
   • Impact: Eliminates ~11K nanoseconds per call from unnecessary dict allocation

2. Smarter Content-Type handling:

   • Original: _headers.setdefault('Content-Type', 'application/json') modifies dict in-place
   • Optimized: if 'Content-Type' not in _headers: _headers = {**_headers, 'Content-Type': 'application/json'} only creates new dict when needed
   • Impact: Reduces dict operations when Content-Type already exists

3. Simplified condition checking:

   • Original: if 'body_additional' in locals() and body_additional: uses expensive locals() lookup
   • Optimized: if body_additional: relies on direct variable check
   • Impact: Eliminates ~237K nanoseconds from locals() introspection

4. Optimized header merging in HTTPClient:

   • Original: {**self.headers, **request.headers} always creates merged dict
   • Optimized: self.headers if not request.headers else {**self.headers, **request.headers} avoids merge when request.headers is empty
   • Impact: Reduces allocations for requests without custom headers

Performance Profile: These optimizations are particularly effective for:

• High-frequency API calls where the function is called repeatedly
• Scenarios with minimal custom headers (most common case)
• Large-scale concurrent operations where small per-call savings compound significantly

The optimizations maintain identical behavior while reducing CPU overhead through smarter conditional logic and reduced object allocations, making the code more efficient for typical JIRA API usage patterns.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio # Used for async execution and concurrency

import pytest # Used for unit testing
from app.sources.external.jira.jira import JiraDataSource

--- Minimal stubs for dependencies (no external libraries except pytest) ---

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_data):
self.response_data = response_data

--- Mock Jira client and its HTTP client ---

class MockHTTPClient:
def init(self, base_url='https://mockjira.com'):
self._base_url = base_url
self.executed_requests = []

def get_base_url(self):
    return self._base_url

async def execute(self, request, **kwargs):
    # Simulate a response with the request data for inspection
    self.executed_requests.append(request)
    # Simulate response data for testing
    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

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

--- Unit Tests ---

1. Basic Test Cases

@pytest.mark.asyncio
async def test_set_default_levels_basic_returns_expected_response():
"""Test basic async/await behavior and that the function returns expected values."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
default_values = [{'id': 1, 'level': 'high'}]
response = await datasource.set_default_levels(default_values)

@pytest.mark.asyncio
async def test_set_default_levels_basic_with_headers_and_body_additional():
"""Test with custom headers and additional body fields."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
default_values = [{'id': 2, 'level': 'medium'}]
headers = {'X-Custom-Header': 'value'}
body_additional = {'extra': 'data'}
response = await datasource.set_default_levels(default_values, body_additional, headers)

@pytest.mark.asyncio
async def test_set_default_levels_basic_empty_default_values():
"""Test with empty defaultValues list."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
default_values = []
response = await datasource.set_default_levels(default_values)

2. Edge Test Cases

@pytest.mark.asyncio
async def test_set_default_levels_raises_if_client_is_none():
"""Test that ValueError is raised if HTTP client is not initialized."""
class DummyClient:
def get_client(self):
return None
jira_client = DummyClient()
with pytest.raises(ValueError, match='HTTP client is not initialized'):
JiraDataSource(jira_client)

@pytest.mark.asyncio
async def test_set_default_levels_raises_if_client_missing_get_base_url():
"""Test that ValueError is raised if HTTP client lacks get_base_url."""
class NoBaseUrlClient:
pass
class DummyClient:
def get_client(self):
return NoBaseUrlClient()
jira_client = DummyClient()
with pytest.raises(ValueError, match='HTTP client does not have get_base_url method'):
JiraDataSource(jira_client)

@pytest.mark.asyncio
async def test_set_default_levels_concurrent_execution():
"""Test concurrent execution of the async function."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
default_values_list = [
[{'id': i, 'level': f'level_{i}'}] for i in range(5)
]
# Run 5 concurrent calls
tasks = [
datasource.set_default_levels(default_values)
for default_values in default_values_list
]
responses = await asyncio.gather(*tasks)
# Each response should correspond to its input
for i, response in enumerate(responses):
pass

@pytest.mark.asyncio
async def test_set_default_levels_handles_special_characters_in_body():
"""Test with special characters and unicode in defaultValues."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
default_values = [{'id': 3, 'level': '特殊字符', 'desc': 'emoji 🚀'}]
response = await datasource.set_default_levels(default_values)

@pytest.mark.asyncio
async def test_set_default_levels_with_none_body_additional_and_headers():
"""Test passing None for body_additional and headers."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
default_values = [{'id': 4, 'level': 'low'}]
response = await datasource.set_default_levels(default_values, None, None)

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_set_default_levels_large_scale_many_requests():
"""Test scalability with many concurrent requests (up to 100)."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
# Generate 100 different defaultValues payloads
default_values_list = [
[{'id': i, 'level': f'level_{i}'}] for i in range(100)
]
tasks = [
datasource.set_default_levels(default_values)
for default_values in default_values_list
]
responses = await asyncio.gather(*tasks)
# Check that each response matches its input
for i, response in enumerate(responses):
pass

@pytest.mark.asyncio
async def test_set_default_levels_large_scale_payload_size():
"""Test with a large defaultValues payload (500 items)."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
large_default_values = [{'id': i, 'level': f'level_{i}'} for i in range(500)]
response = await datasource.set_default_levels(large_default_values)

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_set_default_levels_throughput_small_load():
"""Throughput test: small load (10 requests)."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
default_values_list = [
[{'id': i, 'level': f'small_{i}'}] for i in range(10)
]
tasks = [
datasource.set_default_levels(default_values)
for default_values in default_values_list
]
responses = await asyncio.gather(*tasks)
for i, response in enumerate(responses):
pass

@pytest.mark.asyncio
async def test_set_default_levels_throughput_medium_load():
"""Throughput test: medium load (50 requests)."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
default_values_list = [
[{'id': i, 'level': f'medium_{i}'}] for i in range(50)
]
tasks = [
datasource.set_default_levels(default_values)
for default_values in default_values_list
]
responses = await asyncio.gather(*tasks)
for i, response in enumerate(responses):
pass

@pytest.mark.asyncio
async def test_set_default_levels_throughput_high_volume():
"""Throughput test: high volume (200 requests)."""
mock_client = MockHTTPClient()
jira_client = JiraClient(mock_client)
datasource = JiraDataSource(jira_client)
default_values_list = [
[{'id': i, 'level': f'high_{i}'}] for i in range(200)
]
tasks = [
datasource.set_default_levels(default_values)
for default_values in default_values_list
]
responses = await asyncio.gather(*tasks)
for i, response in enumerate(responses):
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 dependencies ----

These are minimal implementations to allow the tests to run.

They simulate the behavior of the real classes/interfaces.

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_data):
self.data = response_data

class DummyHTTPClient:
"""A dummy async HTTP client for testing."""
def init(self, base_url="http://testserver"):
self._base_url = base_url
self.executed_requests = []

def get_base_url(self):
    return self._base_url

async def execute(self, request, **kwargs):
    # Simulate a successful HTTPResponse
    self.executed_requests.append(request)
    # For testing, echo back the request body in the response
    return HTTPResponse({
        "method": request.method,
        "url": request.url,
        "headers": request.headers,
        "body": request.body,
    })

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

def get_client(self):
    return self.client

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

---- Test cases for JiraDataSource.set_default_levels ----

1. Basic Test Cases

@pytest.mark.asyncio
async def test_set_default_levels_basic_success():
"""Test basic successful call with minimal valid input."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
default_values = [{"level": "admin", "id": 1}]
resp = await ds.set_default_levels(default_values)

@pytest.mark.asyncio
async def test_set_default_levels_with_headers_and_body_additional():
"""Test passing custom headers and additional body properties."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
default_values = [{"level": "user", "id": 2}]
headers = {"Authorization": "Bearer testtoken"}
body_additional = {"extra_field": "extra_value"}
resp = await ds.set_default_levels(default_values, body_additional=body_additional, headers=headers)

@pytest.mark.asyncio
async def test_set_default_levels_empty_default_values():
"""Test with empty defaultValues list."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
default_values = []
resp = await ds.set_default_levels(default_values)

2. Edge Test Cases

@pytest.mark.asyncio
async def test_set_default_levels_client_none_raises():
"""Test that ValueError is raised when client is None."""
class BadClient:
def get_client(self):
return None
with pytest.raises(ValueError, match="HTTP client is not initialized"):
JiraDataSource(BadClient())

@pytest.mark.asyncio
async def test_set_default_levels_client_missing_get_base_url_raises():
"""Test that ValueError is raised 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_set_default_levels_concurrent_execution():
"""Test concurrent execution of set_default_levels."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
default_values_1 = [{"level": "admin", "id": 1}]
default_values_2 = [{"level": "user", "id": 2}]
# Run two calls concurrently
results = await asyncio.gather(
ds.set_default_levels(default_values_1),
ds.set_default_levels(default_values_2)
)

@pytest.mark.asyncio
async def test_set_default_levels_body_additional_overrides_default_values():
"""Test that body_additional can override keys in the main body."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
default_values = [{"level": "admin", "id": 1}]
body_additional = {"defaultValues": [{"level": "user", "id": 2}]}
resp = await ds.set_default_levels(default_values, body_additional=body_additional)

@pytest.mark.asyncio
async def test_set_default_levels_headers_content_type_override():
"""Test that custom Content-Type header is used if provided."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
default_values = [{"level": "admin", "id": 1}]
headers = {"Content-Type": "application/custom-json"}
resp = await ds.set_default_levels(default_values, headers=headers)

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_set_default_levels_large_default_values():
"""Test with a large number of defaultValues (up to 500)."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
default_values = [{"level": "bulk", "id": i} for i in range(500)]
resp = await ds.set_default_levels(default_values)

@pytest.mark.asyncio
async def test_set_default_levels_concurrent_large_scale():
"""Test concurrent execution with multiple large requests."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
requests = [
ds.set_default_levels([{"level": "bulk", "id": i} for i in range(100)])
for _ in range(10)
]
results = await asyncio.gather(*requests)
for resp in results:
pass

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_set_default_levels_throughput_small_load():
"""Throughput: Test performance with small load (10 sequential calls)."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
for i in range(10):
default_values = [{"level": "user", "id": i}]
resp = await ds.set_default_levels(default_values)

@pytest.mark.asyncio
async def test_set_default_levels_throughput_medium_concurrent_load():
"""Throughput: Test concurrent load with medium volume (50 calls)."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
calls = [
ds.set_default_levels([{"level": "concurrent", "id": i}])
for i in range(50)
]
results = await asyncio.gather(*calls)
# Check all responses are correct
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_set_default_levels_throughput_high_volume():
"""Throughput: Test high volume concurrent calls (100 calls)."""
client = JiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
calls = [
ds.set_default_levels([{"level": "high", "id": i}])
for i in range(100)
]
results = await asyncio.gather(*calls)
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.set_default_levels-mhrsxv12 and push.