⚡️ Speed up method JiraDataSource.get_issue_type_scheme_for_projects by 13%

[codeflash-ai]

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

⏱️ Runtime : 2.40 milliseconds → 2.12 milliseconds (best of 231 runs)

📝 Explanation and details

The optimization achieves a 13% runtime improvement and 1.8% throughput increase through targeted micro-optimizations that reduce unnecessary object allocations and string conversions.

Key Optimizations Applied:

1. Fast-path for empty dictionaries - Added early returns in _as_str_dict and _safe_format_url when dictionaries are empty, avoiding expensive dictionary comprehensions and string formatting operations entirely.

2. Reduced dictionary allocations - Changed _headers: Dict[str, Any] = dict(headers or {}) to _headers = headers if headers else {}, eliminating unnecessary dict() constructor calls when headers are already provided or when using empty defaults.

3. Optimized header merging in HTTPClient - Added conditional logic to avoid dictionary unpacking operations ({**self.headers, **request.headers}) when either dictionary is empty, reducing allocation overhead.

4. Streamlined value serialization - Implemented efficient _serialize_value and _to_bool_str functions that handle type-specific conversions with minimal overhead, particularly for boolean values and collections.

Performance Impact Analysis:

From the line profiler results, the most significant gains come from:

• _safe_format_url: Reduced from 596μs to 182μs (69% faster) by short-circuiting empty parameter dictionaries
• _as_str_dict: Improved from 3.87ms to 3.43ms (11% faster) through early empty-dict detection
• Overall method execution: Reduced from 12.4ms to 11.5ms

Test Case Effectiveness:
The optimizations are particularly effective for:

• Basic usage patterns with minimal or no optional parameters (most common case)
• Edge cases with empty project lists or missing headers
• Throughput scenarios with repeated calls, where the cumulative effect of reduced allocations becomes significant

These micro-optimizations target the most frequently executed code paths without changing the API or functionality, making them safe for production deployment while providing measurable performance gains.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio

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

--- Minimal stub classes to allow testing without external dependencies ---

class HTTPResponse:
"""Stub for HTTPResponse returned by HTTPClient.execute"""
def init(self, content, status_code=200):
self.content = content
self.status_code = status_code

def json(self):
    # Simulate JSON response
    return self.content

class HTTPRequest:
"""Stub for HTTPRequest used in HTTPClient.execute"""
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 stub JiraClient and HTTPClient for testing ---

class StubHTTPClient:
"""Stub HTTPClient for async execute method"""
def init(self):
self._last_request = None
self._should_raise = False
self._response_content = {"schemes": ["default"]}
self._response_status = 200

def get_base_url(self):
    return "https://jira.example.com"

async def execute(self, request, **kwargs):
    self._last_request = request
    if self._should_raise:
        raise RuntimeError("Simulated network error")
    # Simulate response based on query params
    return HTTPResponse(
        {
            "method": request.method,
            "url": request.url,
            "headers": request.headers,
            "query_params": request.query_params,
            "body": request.body,
            "schemes": self._response_content.get("schemes", []),
            "projectId": request.query_params.get("projectId"),
            "startAt": request.query_params.get("startAt"),
            "maxResults": request.query_params.get("maxResults"),
        },
        status_code=self._response_status
    )

class JiraRESTClientViaApiKey(StubHTTPClient):
pass

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_basic_returns_expected_values():
"""Basic: Function returns expected HTTPResponse with correct params"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
project_ids = [101, 202]
response = await ds.get_issue_type_scheme_for_projects(projectId=project_ids)

@pytest.mark.asyncio
async def test_basic_with_optional_params():
"""Basic: Function correctly handles optional startAt and maxResults"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
response = await ds.get_issue_type_scheme_for_projects(
projectId=[1, 2, 3],
startAt=10,
maxResults=50
)

@pytest.mark.asyncio
async def test_basic_with_headers():
"""Basic: Function correctly accepts custom headers"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
custom_headers = {"X-Test-Header": "test-value"}
response = await ds.get_issue_type_scheme_for_projects(
projectId=[99],
headers=custom_headers
)

@pytest.mark.asyncio
async def test_edge_empty_project_id_list():
"""Edge: Handles empty projectId list"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
response = await ds.get_issue_type_scheme_for_projects(projectId=[])

@pytest.mark.asyncio
async def test_edge_project_id_single_element():
"""Edge: Handles projectId with one element"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
response = await ds.get_issue_type_scheme_for_projects(projectId=[42])

@pytest.mark.asyncio
async def test_edge_project_id_non_int_values():
"""Edge: Handles projectId list with non-int values (should serialize as str)"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
response = await ds.get_issue_type_scheme_for_projects(projectId=["A", "B"])

@pytest.mark.asyncio
async def test_edge_none_client_raises():
"""Edge: Raises ValueError if HTTP client is None"""
class BadClient:
def get_client(self):
return None
with pytest.raises(ValueError) as excinfo:
JiraDataSource(BadClient())

@pytest.mark.asyncio
async def test_edge_missing_get_base_url_raises():
"""Edge: Raises ValueError if client lacks get_base_url"""
class BadHTTPClient:
pass
class BadClient:
def get_client(self):
return BadHTTPClient()
with pytest.raises(ValueError) as excinfo:
JiraDataSource(BadClient())

@pytest.mark.asyncio
async def test_edge_execute_raises_exception():
"""Edge: Handles exception from HTTPClient.execute"""
client = JiraRESTClientViaApiKey()
client._should_raise = True
ds = JiraDataSource(JiraClient(client))
with pytest.raises(RuntimeError) as excinfo:
await ds.get_issue_type_scheme_for_projects(projectId=[1])

@pytest.mark.asyncio
async def test_concurrent_execution():
"""Edge: Multiple concurrent requests return correct results"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
async def call(project_id):
return await ds.get_issue_type_scheme_for_projects(projectId=[project_id])
results = await asyncio.gather(
call(111),
call(222),
call(333)
)

@pytest.mark.asyncio
async def test_large_scale_many_project_ids():
"""Large scale: Handles large projectId list (<=1000 elements)"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
large_list = list(range(1, 1001))
response = await ds.get_issue_type_scheme_for_projects(projectId=large_list)

@pytest.mark.asyncio
async def test_large_scale_concurrent_calls():
"""Large scale: Handles multiple concurrent calls"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
async def call(idx):
return await ds.get_issue_type_scheme_for_projects(projectId=[idx])
tasks = [call(i) for i in range(25)]
results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_JiraDataSource_get_issue_type_scheme_for_projects_throughput_small_load():
"""Throughput: Small load, repeated calls"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
for i in range(10):
response = await ds.get_issue_type_scheme_for_projects(projectId=[i])

@pytest.mark.asyncio
async def test_JiraDataSource_get_issue_type_scheme_for_projects_throughput_medium_load():
"""Throughput: Medium load, concurrent calls"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
async def call(idx):
return await ds.get_issue_type_scheme_for_projects(projectId=[idx])
tasks = [call(i) for i in range(50)]
results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_JiraDataSource_get_issue_type_scheme_for_projects_throughput_large_load():
"""Throughput: Large load, concurrent calls (<=100)"""
client = JiraClient(JiraRESTClientViaApiKey())
ds = JiraDataSource(client)
async def call(idx):
return await ds.get_issue_type_scheme_for_projects(projectId=[idx])
tasks = [call(i) for i in range(100)]
results = await asyncio.gather(*tasks)

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 required classes ----

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):
    # Simulate a json response for test purposes
    return self.response

---- Mock JiraClient and HTTPClient ----

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: HTTPRequest):
    # Simulate response based on query params for test purposes
    # Return a dict with the query_params and headers for verification
    self.executed_requests.append(request)
    # Simulate various responses based on input for edge/throughput tests
    response = {
        "url": request.url,
        "headers": request.headers,
        "query_params": request.query_params,
        "body": request.body,
        "method": request.method,
    }
    # Simulate error for specific test case
    if request.query_params.get("projectId") == ["error"]:
        raise ValueError("Simulated error in execute")
    return HTTPResponse(response)

class JiraClient:
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_basic_single_project_id():
"""Test with a single valid projectId and default parameters."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
resp = await ds.get_issue_type_scheme_for_projects([123])

@pytest.mark.asyncio
async def test_basic_multiple_project_ids():
"""Test with multiple projectIds."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
resp = await ds.get_issue_type_scheme_for_projects([1, 2, 3])

@pytest.mark.asyncio
async def test_basic_with_startAt_and_maxResults():
"""Test with startAt and maxResults provided."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
resp = await ds.get_issue_type_scheme_for_projects([42], startAt=10, maxResults=25)
qp = resp.json()["query_params"]

@pytest.mark.asyncio
async def test_basic_with_custom_headers():
"""Test with custom headers provided."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
headers = {"X-Test-Header": "test-value"}
resp = await ds.get_issue_type_scheme_for_projects([99], headers=headers)

2. Edge Test Cases

@pytest.mark.asyncio
async def test_edge_empty_project_id_list():
"""Test with an empty projectId list."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
resp = await ds.get_issue_type_scheme_for_projects([])

@pytest.mark.asyncio
async def test_edge_project_id_with_zero_and_negative():
"""Test with projectId containing zero and negative values."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
resp = await ds.get_issue_type_scheme_for_projects([0, -1, 2])

@pytest.mark.asyncio
async def test_edge_large_project_id_numbers():
"""Test with large integer projectIds."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
large_ids = [999999999, 888888888]
resp = await ds.get_issue_type_scheme_for_projects(large_ids)

@pytest.mark.asyncio
async def test_edge_headers_with_non_string_values():
"""Test with headers containing non-string values."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
headers = {"X-Int": 123, "X-Bool": True}
resp = await ds.get_issue_type_scheme_for_projects([1], headers=headers)

@pytest.mark.asyncio
async def test_edge_client_not_initialized():
"""Test ValueError if client is not initialized."""
class BadClient:
def get_client(self): return None
with pytest.raises(ValueError):
JiraDataSource(BadClient())

@pytest.mark.asyncio
async def test_edge_client_missing_get_base_url():
"""Test ValueError if client lacks get_base_url method."""
class NoBaseUrlClient:
pass
class Wrapper:
def get_client(self): return NoBaseUrlClient()
with pytest.raises(ValueError):
JiraDataSource(Wrapper())

@pytest.mark.asyncio
async def test_edge_execute_raises_exception():
"""Test that exceptions in execute are propagated."""
class ErrorHTTPClient(MockHTTPClient):
async def execute(self, request):
raise RuntimeError("Simulated HTTP error")
client = JiraClient(ErrorHTTPClient())
ds = JiraDataSource(client)
with pytest.raises(RuntimeError):
await ds.get_issue_type_scheme_for_projects([1])

@pytest.mark.asyncio

async def test_edge_concurrent_requests():
"""Test concurrent execution of multiple requests."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
# Run 5 requests concurrently
coros = [
ds.get_issue_type_scheme_for_projects([i], startAt=i*10, maxResults=5)
for i in range(5)
]
results = await asyncio.gather(*coros)
for idx, resp in enumerate(results):
qp = resp.json()["query_params"]

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_large_scale_many_project_ids():
"""Test with a large number of projectIds (up to 1000)."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
ids = list(range(1, 1001))
resp = await ds.get_issue_type_scheme_for_projects(ids)
# Should serialize to comma-separated string
expected = ",".join(str(i) for i in ids)

@pytest.mark.asyncio
async def test_large_scale_many_concurrent_requests():
"""Test with many concurrent requests (up to 100)."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
coros = [
ds.get_issue_type_scheme_for_projects([i, i+1])
for i in range(0, 100, 2)
]
results = await asyncio.gather(coros)
for idx, resp in enumerate(results):
ids = [idx2, idx*2+1]
expected = ",".join(str(i) for i in ids)

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_JiraDataSource_get_issue_type_scheme_for_projects_throughput_small_load():
"""Throughput test: small load, 10 concurrent requests."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
coros = [
ds.get_issue_type_scheme_for_projects([i])
for i in range(10)
]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_JiraDataSource_get_issue_type_scheme_for_projects_throughput_medium_load():
"""Throughput test: medium load, 50 concurrent requests."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
coros = [
ds.get_issue_type_scheme_for_projects([i])
for i in range(50)
]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_JiraDataSource_get_issue_type_scheme_for_projects_throughput_large_load():
"""Throughput test: large load, 200 concurrent requests."""
client = JiraClient(MockHTTPClient())
ds = JiraDataSource(client)
coros = [
ds.get_issue_type_scheme_for_projects([i])
for i in range(200)
]
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.

To edit these changes git checkout codeflash/optimize-JiraDataSource.get_issue_type_scheme_for_projects-mhs6asj4 and push.