⚡️ Speed up method JiraDataSource.get_alternative_issue_types by 29%

[codeflash-ai]

📄 29% (0.29x) speedup for JiraDataSource.get_alternative_issue_types in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 1.73 milliseconds → 1.34 milliseconds (best of 43 runs)

📝 Explanation and details

The optimization achieves a 29% runtime improvement (1.73ms → 1.34ms) and 2.4% throughput improvement (19,320 → 19,780 ops/sec) by eliminating unnecessary dictionary allocations and function calls in the HTTP request construction path.

Key optimizations applied:

1. Eliminated redundant dictionary creation: Removed intermediate _headers, _path, _query, and _body variables, directly constructing the HTTPRequest with inline values. This saves ~4 dictionary allocations per call.

2. Optimized string conversion with _fast_as_str_dict: The new helper includes fast paths for empty dictionaries (return {}) and single-item dictionaries (avoiding comprehension overhead). Since headers are often None or small, this significantly reduces conversion costs.

3. Direct query params assignment: Since query_params is always empty for this endpoint, it's hardcoded as {} instead of calling _as_str_dict({}), eliminating one function call per request.

4. Improved _safe_format_url: Added an early check for empty params to skip _SafeDict creation when unnecessary.

Performance impact analysis:

• Line profiler shows the most expensive operations (_as_str_dict calls) dropped from 45% of total time to much less
• The HTTPRequest construction time reduced from 15.7% to 25.6% of total time, but with much lower absolute cost
• URL formatting became slightly more expensive (19.2% → 37.7%) but still faster in absolute terms

Workload suitability:
These optimizations are particularly effective for:

• High-frequency API calls with minimal or no headers
• Concurrent request patterns (as shown in throughput tests with 50-100 concurrent requests)
• Sustained load scenarios where the reduced allocation overhead compounds

The changes maintain full API compatibility while delivering consistent performance gains across all test scenarios, from basic single requests to high-volume concurrent workloads.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 490 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 dependencies ----

class HTTPResponse:
"""Stub for HTTPResponse, simulates a response object."""
def init(self, data):
self.data = data

class HTTPRequest:
"""Stub for HTTPRequest, simulates an HTTP request object."""
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

---- Minimal JiraClient and HTTPClient stubs ----

class DummyHTTPClient:
"""Stub for HTTP client used by JiraDataSource."""
def init(self, base_url='https://jira.example.com'):
self._base_url = base_url
self.executed_requests = []

def get_base_url(self):
    return self._base_url

async def execute(self, request):
    # Record the request for test assertions
    self.executed_requests.append(request)
    # Simulate a response containing the request info for easy verification
    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 DummyJiraClient:
"""Stub for JiraClient, returns DummyHTTPClient."""
def init(self, client):
self.client = client

def get_client(self):
    return self.client

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

---- Basic Test Cases ----

@pytest.mark.asyncio
async def test_get_alternative_issue_types_basic_returns_expected_response():
"""Test basic async invocation returns expected HTTPResponse data."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
response = await ds.get_alternative_issue_types('123')

@pytest.mark.asyncio
async def test_get_alternative_issue_types_basic_with_headers():
"""Test that headers are passed and serialized correctly."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
headers = {'Authorization': 'Bearer token', 'Custom-Header': 42}
response = await ds.get_alternative_issue_types('abc', headers)

@pytest.mark.asyncio
async def test_get_alternative_issue_types_basic_async_await_behavior():
"""Test that the function returns a coroutine and can be awaited."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
# Should be awaitable and return HTTPResponse
codeflash_output = ds.get_alternative_issue_types('xyz'); coro = codeflash_output
result = await coro

---- Edge Test Cases ----

@pytest.mark.asyncio
async def test_get_alternative_issue_types_concurrent_execution():
"""Test concurrent execution of the async function with different ids."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
ids = ['a', 'b', 'c', 'd']
coros = [ds.get_alternative_issue_types(i) for i in ids]
results = await asyncio.gather(*coros)
# Each result should have the correct id in the path_params
for idx, result in enumerate(results):
pass

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

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

@pytest.mark.asyncio
async def test_get_alternative_issue_types_empty_id():
"""Test edge case with empty string as id."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
response = await ds.get_alternative_issue_types('')

@pytest.mark.asyncio
async def test_get_alternative_issue_types_special_characters_id():
"""Test edge case with special characters in id."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
special_id = 'id/with/slash?and=equals'
response = await ds.get_alternative_issue_types(special_id)

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

@pytest.mark.asyncio
async def test_get_alternative_issue_types_many_concurrent_requests():
"""Test large scale concurrent execution with many requests."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
ids = [f"id_{i}" for i in range(50)] # 50 concurrent requests
coros = [ds.get_alternative_issue_types(i) for i in ids]
results = await asyncio.gather(*coros)
# All results should be correct and unique
for idx, result in enumerate(results):
pass

@pytest.mark.asyncio

async def test_get_alternative_issue_types_throughput_small_load():
"""Throughput test: small load of 5 concurrent requests."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
ids = [str(i) for i in range(5)]
coros = [ds.get_alternative_issue_types(i) for i in ids]
results = await asyncio.gather(*coros)
for idx, result in enumerate(results):
pass

@pytest.mark.asyncio
async def test_get_alternative_issue_types_throughput_medium_load():
"""Throughput test: medium load of 20 concurrent requests."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
ids = [str(i) for i in range(20)]
coros = [ds.get_alternative_issue_types(i) for i in ids]
results = await asyncio.gather(*coros)
for idx, result in enumerate(results):
pass

@pytest.mark.asyncio
async def test_get_alternative_issue_types_throughput_high_volume():
"""Throughput test: high volume of 100 concurrent requests."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
ids = [str(i) for i in range(100)]
coros = [ds.get_alternative_issue_types(i) for i in ids]
results = await asyncio.gather(*coros)
for idx, result in enumerate(results):
pass

@pytest.mark.asyncio
async def test_get_alternative_issue_types_throughput_sustained_pattern():
"""Throughput test: sustained execution pattern, repeated calls."""
client = DummyHTTPClient()
jira_client = DummyJiraClient(client)
ds = JiraDataSource(jira_client)
ids = [f"repeat_{i}" for i in range(10)]
# Repeat the pattern 5 times for sustained load
for _ in range(5):
coros = [ds.get_alternative_issue_types(i) for i in ids]
results = await asyncio.gather(*coros)
for idx, result 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

--- Helpers from the original code (copied as-is) ---

from typing import Any, Dict, Optional, Union

import pytest # used for our unit tests
from app.sources.external.jira.jira import JiraDataSource

--- Minimal stubs for dependencies ---

class DummyHTTPResponse:
"""A dummy HTTPResponse to simulate real HTTP responses."""
def init(self, data=None):
self.data = data or {}
def eq(self, other):
return isinstance(other, DummyHTTPResponse) and self.data == other.data
def repr(self):
return f"DummyHTTPResponse({self.data})"

class DummyHTTPRequest:
"""A dummy HTTPRequest to capture request construction."""
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 DummyClient:
"""A dummy client to simulate the Jira client."""
def init(self, base_url, should_raise=False, execute_result=None, execute_delay=0):
self._base_url = base_url
self.should_raise = should_raise
self.execute_result = execute_result or DummyHTTPResponse({"ok": True})
self.execute_delay = execute_delay
self.last_request = None # for inspection in tests
def get_base_url(self):
if self.should_raise:
raise AttributeError("No base url")
return self._base_url
async def execute(self, req):
self.last_request = req
if self.should_raise:
raise ValueError("Execution failed")
if self.execute_delay > 0:
await asyncio.sleep(self.execute_delay)
return self.execute_result

class DummyJiraClient:
"""A dummy JiraClient wrapper."""
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_get_alternative_issue_types_basic_returns_expected_response():
"""Test that the function returns the expected DummyHTTPResponse for a normal id."""
client = DummyClient(base_url="https://jira.example.com")
ds = JiraDataSource(DummyJiraClient(client))
result = await ds.get_alternative_issue_types("123")

@pytest.mark.asyncio
async def test_get_alternative_issue_types_basic_url_and_headers():
"""Test that the function constructs the correct URL and headers."""
client = DummyClient(base_url="https://jira.example.com")
ds = JiraDataSource(DummyJiraClient(client))
headers = {"X-Test": "abc", "Content-Type": "application/json"}
await ds.get_alternative_issue_types("456", headers=headers)
# Check the last_request on the client
req = client.last_request

@pytest.mark.asyncio
async def test_get_alternative_issue_types_basic_empty_headers():
"""Test that the function handles empty headers gracefully."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
await ds.get_alternative_issue_types("789", headers={})
req = client.last_request

@pytest.mark.asyncio
async def test_get_alternative_issue_types_basic_default_headers_none():
"""Test that the function works when headers=None (the default)."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
await ds.get_alternative_issue_types("321")
req = client.last_request

@pytest.mark.asyncio
async def test_get_alternative_issue_types_basic_id_special_characters():
"""Test that the function correctly inserts special characters in the id."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
special_id = "abc-!@#%$^&*()_+"
await ds.get_alternative_issue_types(special_id)
req = client.last_request

2. Edge Test Cases

@pytest.mark.asyncio
async def test_get_alternative_issue_types_edge_none_client_raises():
"""Test that ValueError is raised if the JiraClient returns None for get_client()."""
class NoneJiraClient:
def get_client(self):
return None
with pytest.raises(ValueError, match="HTTP client is not initialized"):
JiraDataSource(NoneJiraClient())

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

@pytest.mark.asyncio

async def test_get_alternative_issue_types_edge_empty_id():
"""Test that the function works when id is an empty string."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
await ds.get_alternative_issue_types("")
req = client.last_request

@pytest.mark.asyncio
async def test_get_alternative_issue_types_edge_headers_with_non_str_values():
"""Test that non-string header values are converted to strings."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
headers = {"X-Int": 123, "X-Bool": True, "X-Float": 3.14}
await ds.get_alternative_issue_types("id", headers=headers)
req = client.last_request

@pytest.mark.asyncio
async def test_get_alternative_issue_types_edge_path_params_non_str():
"""Test that non-string path param values are stringified."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
await ds.get_alternative_issue_types(999) # id as int
req = client.last_request

@pytest.mark.asyncio
async def test_get_alternative_issue_types_edge_concurrent_calls():
"""Test that concurrent calls do not interfere with each other."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
async def call(id_val):
return await ds.get_alternative_issue_types(id_val)
results = await asyncio.gather(
call("id1"),
call("id2"),
call("id3"),
)
for r in results:
pass

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_get_alternative_issue_types_large_scale_many_concurrent():
"""Test a moderate number of concurrent calls for scalability."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
ids = [f"bulk-{i}" for i in range(50)]
async def call(id_val):
return await ds.get_alternative_issue_types(id_val)
results = await asyncio.gather(*(call(i) for i in ids))
for r in results:
pass

@pytest.mark.asyncio
async def test_get_alternative_issue_types_large_scale_unique_headers():
"""Test that each concurrent call can accept unique headers."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
async def call(idx):
headers = {"Call-Id": idx}
return await ds.get_alternative_issue_types(f"id-{idx}", headers=headers)
results = await asyncio.gather(*(call(i) for i in range(10)))

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_get_alternative_issue_types_throughput_small_load():
"""Test throughput for a small batch of requests."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
ids = [f"tps-{i}" for i in range(5)]
async def call(id_val):
return await ds.get_alternative_issue_types(id_val)
results = await asyncio.gather(*(call(i) for i in ids))

@pytest.mark.asyncio
async def test_get_alternative_issue_types_throughput_medium_load():
"""Test throughput for a medium batch of requests."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
ids = [f"tpm-{i}" for i in range(30)]
async def call(id_val):
return await ds.get_alternative_issue_types(id_val)
results = await asyncio.gather(*(call(i) for i in ids))

@pytest.mark.asyncio
async def test_get_alternative_issue_types_throughput_high_volume():
"""Test throughput for a high volume of requests (but <1000)."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
ids = [f"tph-{i}" for i in range(100)]
async def call(id_val):
return await ds.get_alternative_issue_types(id_val)
results = await asyncio.gather(*(call(i) for i in ids))

@pytest.mark.asyncio
async def test_get_alternative_issue_types_throughput_sustained_pattern():
"""Test throughput with repeated sequential calls to simulate sustained load."""
client = DummyClient(base_url="http://jira.local")
ds = JiraDataSource(DummyJiraClient(client))
for i in range(20):
result = await ds.get_alternative_issue_types(f"sustain-{i}")

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.get_alternative_issue_types-mhs1vusa and push.