⚡️ Speed up method JiraDataSource.update_issue_type_scheme by 5%

[codeflash-ai]

📄 5% (0.05x) speedup for JiraDataSource.update_issue_type_scheme in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 2.07 milliseconds → 1.97 milliseconds (best of 33 runs)

📝 Explanation and details

The optimized code achieves a 5% runtime improvement and 3.1% throughput improvement through strategic micro-optimizations that reduce object allocations and attribute lookups in the hot path of the update_issue_type_scheme method.

Key optimizations applied:

1. Client caching: client = self._client caches the client reference locally, eliminating repeated self._client attribute lookups (saving ~475 lookups per call based on line profiler data).

2. Conditional dict construction: Changed dict(headers or {}) to dict(headers) if headers else {}, avoiding unnecessary empty dict creation when headers is None.

3. Direct empty dict assignment: Replaced _query: Dict[str, Any] = {} with direct query_params={} in HTTPRequest constructor, eliminating one dict allocation and variable assignment.

4. Reduced _as_str_dict calls: The line profiler shows _as_str_dict was called 1425 times in the original vs 950 times in the optimized version, indicating elimination of one unnecessary conversion (the empty query dict).

Performance impact:

• The optimizations target the method's setup overhead before the actual HTTP request
• Line profiler shows the total function time decreased from 10.98ms to 9.96ms (9% improvement in CPU-bound portions)
• The _as_str_dict function shows 17% fewer total calls, reducing string conversion overhead
• These micro-optimizations are particularly beneficial for high-throughput scenarios where this method is called repeatedly

Test case performance:
Based on the annotated tests, the optimizations show consistent improvements across all workload types - basic calls, edge cases, and especially throughput tests with 20-100 concurrent requests where the reduced per-call overhead compounds significantly.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 433 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 used by JiraDataSource ---

HTTPResponse stub for testing

class HTTPResponse:
def init(self, data):
self.data = data

def json(self):
    return self.data

HTTPRequest stub for testing

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

--- Minimal JiraClient and HTTP client stubs for testing ---

class DummyAsyncClient:
def init(self, base_url):
self._base_url = base_url
self.executed_requests = []

def get_base_url(self):
    return self._base_url

async def execute(self, req):
    # Save request for inspection
    self.executed_requests.append(req)
    # Simulate response based on input for testing
    return 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:
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_update_issue_type_scheme_basic_success():
"""Test basic successful update with all parameters provided."""
dummy_client = DummyAsyncClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(dummy_client))
resp = await ds.update_issue_type_scheme(
issueTypeSchemeId=123,
defaultIssueTypeId="10001",
description="Test scheme",
name="Scheme Name"
)

@pytest.mark.asyncio
async def test_update_issue_type_scheme_basic_minimal_params():
"""Test update with only required parameter (issueTypeSchemeId)."""
dummy_client = DummyAsyncClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(dummy_client))
resp = await ds.update_issue_type_scheme(issueTypeSchemeId=456)

@pytest.mark.asyncio
async def test_update_issue_type_scheme_basic_custom_headers():
"""Test update with custom headers merged correctly."""
dummy_client = DummyAsyncClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(dummy_client))
custom_headers = {"X-Test-Header": "foobar"}
resp = await ds.update_issue_type_scheme(
issueTypeSchemeId=789,
headers=custom_headers
)

2. Edge Test Cases

@pytest.mark.asyncio

async def test_update_issue_type_scheme_edge_missing_get_base_url():
"""Test that ValueError is raised if client lacks get_base_url."""
class DummyClientNoBaseUrl:
pass
with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
JiraDataSource(JiraClient(DummyClientNoBaseUrl()))

@pytest.mark.asyncio

async def test_update_issue_type_scheme_edge_body_none_values():
"""Test that None values for optional fields are not included in body."""
dummy_client = DummyAsyncClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(dummy_client))
resp = await ds.update_issue_type_scheme(
issueTypeSchemeId=555,
defaultIssueTypeId=None,
description=None,
name=None
)

3. Large Scale Test Cases

@pytest.mark.asyncio

async def test_update_issue_type_scheme_large_scale_body_fields():
"""Test large body field values (not exceeding 1000 chars)."""
dummy_client = DummyAsyncClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(dummy_client))
long_desc = "x" * 1000
long_name = "y" * 1000
resp = await ds.update_issue_type_scheme(
issueTypeSchemeId=999,
description=long_desc,
name=long_name
)

4. Throughput Test Cases

@pytest.mark.asyncio

async def test_update_issue_type_scheme_throughput_medium_load():
"""Test throughput under medium load (50 requests)."""
dummy_client = DummyAsyncClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(dummy_client))
tasks = [
ds.update_issue_type_scheme(issueTypeSchemeId=i, name=f"Batch{i}")
for i in range(50)
]
results = await asyncio.gather(*tasks)
# All names should be present
names = [resp.data["body"]["name"] for resp in results]

@pytest.mark.asyncio
async def test_update_issue_type_scheme_throughput_high_volume():
"""Test throughput under high volume (100 requests, bounded for speed)."""
dummy_client = DummyAsyncClient("https://jira.example.com")
ds = JiraDataSource(JiraClient(dummy_client))
tasks = [
ds.update_issue_type_scheme(issueTypeSchemeId=i, description=f"desc{i}", name=f"name{i}")
for i in range(100)
]
results = await asyncio.gather(*tasks)
# Check a few sample responses for correctness
for i in [0, 50, 99]:
resp = results[i]

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 HTTPRequest, HTTPResponse, JiraClient, JiraRESTClientViaApiKey ---

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
self.status_code = getattr(response, "status_code", 200)
self.body = getattr(response, "body", {})
self.headers = getattr(response, "headers", {})
self.text = getattr(response, "text", "")

def json(self):
    return self.body

class DummyAsyncResponse:
def init(self, status_code=200, body=None, headers=None, text=""):
self.status_code = status_code
self.body = body if body is not None else {}
self.headers = headers if headers is not None else {}
self.text = text

class DummyAsyncClient:
def init(self, base_url):
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 inspection
    self.executed_requests.append(request)
    # Return a dummy response
    return HTTPResponse(DummyAsyncResponse(
        status_code=200,
        body={
            "received_body": request.body,
            "received_headers": request.headers,
            "received_url": request.url,
            "received_method": request.method,
            "received_path_params": request.path_params,
            "received_query_params": request.query_params,
        },
        headers=request.headers,
        text="OK"
    ))

class JiraRESTClientViaApiKey(DummyAsyncClient):
def init(self, base_url, email, api_key):
super().init(base_url)
self.base_url = base_url

def get_base_url(self):
    return self.base_url

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_update_issue_type_scheme_basic_required_only():
"""Test basic async/await and required param only."""
client = JiraClient(JiraRESTClientViaApiKey("https://jira.example.com", "user@example.com", "apikey"))
ds = JiraDataSource(client)
resp = await ds.update_issue_type_scheme(issueTypeSchemeId=123)

@pytest.mark.asyncio
async def test_update_issue_type_scheme_basic_all_fields():
"""Test all optional fields provided."""
client = JiraClient(JiraRESTClientViaApiKey("https://jira.example.com", "user@example.com", "apikey"))
ds = JiraDataSource(client)
resp = await ds.update_issue_type_scheme(
issueTypeSchemeId=456,
defaultIssueTypeId="issue-type-xyz",
description="A scheme description",
name="Scheme Name"
)
body = resp.json()["received_body"]

@pytest.mark.asyncio
async def test_update_issue_type_scheme_custom_headers():
"""Test custom headers are included and content-type is set."""
client = JiraClient(JiraRESTClientViaApiKey("https://jira.example.com", "user@example.com", "apikey"))
ds = JiraDataSource(client)
custom_headers = {"X-Custom-Header": "value"}
resp = await ds.update_issue_type_scheme(issueTypeSchemeId=789, headers=custom_headers)
headers = resp.json()["received_headers"]

---- 2. Edge Test Cases ----

@pytest.mark.asyncio
async def test_update_issue_type_scheme_invalid_client_raises():
"""Test ValueError raised if 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_update_issue_type_scheme_missing_get_base_url_raises():
"""Test ValueError raised if client has no get_base_url."""
class BadClientObj:
pass
class BadClient:
def get_client(self):
return BadClientObj()
with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
JiraDataSource(BadClient())

@pytest.mark.asyncio
async def test_update_issue_type_scheme_concurrent_execution():
"""Test concurrent async execution returns correct results for each call."""
client = JiraClient(JiraRESTClientViaApiKey("https://jira.example.com", "user@example.com", "apikey"))
ds = JiraDataSource(client)
# Run 5 concurrent updates with different ids
ids = [100, 101, 102, 103, 104]
coros = [ds.update_issue_type_scheme(issueTypeSchemeId=i, name=f"Scheme {i}") for i in ids]
results = await asyncio.gather(*coros)
for i, resp in zip(ids, results):
pass

@pytest.mark.asyncio
async def test_update_issue_type_scheme_with_empty_strings():
"""Test handling of empty string values for optional fields."""
client = JiraClient(JiraRESTClientViaApiKey("https://jira.example.com", "user@example.com", "apikey"))
ds = JiraDataSource(client)
resp = await ds.update_issue_type_scheme(
issueTypeSchemeId=555,
defaultIssueTypeId="",
description="",
name=""
)
body = resp.json()["received_body"]

---- 3. Large Scale Test Cases ----

@pytest.mark.asyncio
async def test_update_issue_type_scheme_many_concurrent_calls():
"""Test many concurrent calls (50) for scalability."""
client = JiraClient(JiraRESTClientViaApiKey("https://jira.example.com", "user@example.com", "apikey"))
ds = JiraDataSource(client)
n_calls = 50
coros = [
ds.update_issue_type_scheme(issueTypeSchemeId=i, name=f"Scheme {i}", description=f"Desc {i}")
for i in range(1000, 1000 + n_calls)
]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
expected_id = str(1000 + i)
body = resp.json()["received_body"]

@pytest.mark.asyncio
async def test_update_issue_type_scheme_varied_inputs():
"""Test a variety of input combinations for robustness."""
client = JiraClient(JiraRESTClientViaApiKey("https://jira.example.com", "user@example.com", "apikey"))
ds = JiraDataSource(client)
# None, empty, and normal values for each optional field
cases = [
dict(issueTypeSchemeId=1),
dict(issueTypeSchemeId=2, defaultIssueTypeId=None, description=None, name=None),
dict(issueTypeSchemeId=3, defaultIssueTypeId="", description="", name=""),
dict(issueTypeSchemeId=4, defaultIssueTypeId="abc", description="desc", name="name"),
dict(issueTypeSchemeId=5, defaultIssueTypeId="xyz", description=None, name="SchemeX"),
]
coros = [ds.update_issue_type_scheme(**case) for case in cases]
results = await asyncio.gather(*coros)
# Validate each result
for case, resp in zip(cases, results):
body = resp.json()["received_body"]
for field in ["defaultIssueTypeId", "description", "name"]:
if field in case and case[field] is not None:
pass
elif case.get(field, None) is None:
pass

---- 4. Throughput Test Cases ----

@pytest.mark.asyncio

async def test_update_issue_type_scheme_throughput_medium_load():
"""Throughput: medium batch of 20 requests."""
client = JiraClient(JiraRESTClientViaApiKey("https://jira.example.com", "user@example.com", "apikey"))
ds = JiraDataSource(client)
coros = [
ds.update_issue_type_scheme(issueTypeSchemeId=i, name=f"Medium {i}")
for i in range(20)
]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_update_issue_type_scheme_throughput_high_load():
"""Throughput: high batch of 100 requests."""
client = JiraClient(JiraRESTClientViaApiKey("https://jira.example.com", "user@example.com", "apikey"))
ds = JiraDataSource(client)
coros = [
ds.update_issue_type_scheme(issueTypeSchemeId=i, name=f"High {i}")
for i in range(100)
]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_update_issue_type_scheme_throughput_varying_loads():
"""Throughput: varying loads in sequence for sustained execution."""
client = JiraClient(JiraRESTClientViaApiKey("https://jira.example.com", "user@example.com", "apikey"))
ds = JiraDataSource(client)
# Small, medium, high batch sizes
batch_sizes = [3, 15, 50]
for size in batch_sizes:
coros = [
ds.update_issue_type_scheme(issueTypeSchemeId=i, name=f"Batch {size} - {i}")
for i in range(size)
]
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.update_issue_type_scheme-mhs7q6uv and push.