⚡️ Speed up method JiraDataSource.get_issue_type_property by 38%

[codeflash-ai]

📄 38% (0.38x) speedup for JiraDataSource.get_issue_type_property in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 3.04 milliseconds → 2.19 milliseconds (best of 231 runs)

📝 Explanation and details

The optimized code achieves a 38% runtime improvement through several targeted micro-optimizations that reduce dictionary allocations and function call overhead:

Key Optimizations Applied:

1. Precomputed Constants: The relative path template _REL_PATH and empty string dictionary _EMPTY_STR_DICT are moved to module level, eliminating repeated string and dict allocations on every function call.

2. Specialized URL Formatting: Replaced the generic _safe_format_url() with _format_issue_type_property_url() that directly formats the known template with two parameters, avoiding the overhead of _SafeDict creation and exception handling.

3. Conditional Header Processing: Headers are only processed when present (3 out of 792 calls in the profiler), avoiding unnecessary dict() and _as_str_dict() calls for the common case of no headers.

4. Optimized Empty Dictionary Handling: The _as_str_dict() function now returns the precomputed _EMPTY_STR_DICT for empty inputs instead of creating new dictionary objects.

5. Reduced Dictionary Allocations: Eliminated creation of intermediate _path, _query, and _body variables, directly using constants and inline values where appropriate.

Performance Impact:

• Runtime improved from 3.04ms to 2.19ms (38% faster)
• Throughput increased from 177,632 to 183,183 operations/second (3.1% improvement)
• Line profiler shows the most expensive operations (_as_str_dict calls on path/query params) dropped from 23.4% + 10.9% = 34.3% of total time to virtually negligible overhead

Test Case Benefits:
The optimizations are particularly effective for:

• High-volume concurrent calls (200+ requests) where dictionary allocation overhead compounds
• Cases with empty/null headers (the common path), which now bypass string conversion entirely
• Repeated calls with similar parameters, benefiting from constant reuse

These micro-optimizations are especially valuable in high-throughput API clients where this method may be called thousands of times per second.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 819 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
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 to allow testing ---

class DummyHTTPResponse:
"""A minimal HTTPResponse stub for test purposes."""
def init(self, status_code=200, data=None):
self.status_code = status_code
self.data = data or {}

def json(self):
    return self.data

class DummyHTTPClient:
"""A dummy async HTTP client that simulates HTTP requests."""
def init(self, base_url="https://dummy.atlassian.net"):
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)
    # Simulate different responses based on input for edge testing
    if request.path_params.get("issueTypeId") == "notfound":
        return DummyHTTPResponse(status_code=404, data={"error": "Not Found"})
    if request.path_params.get("propertyKey") == "error":
        raise RuntimeError("Simulated client error")
    return DummyHTTPResponse(status_code=200, data={
        "issueTypeId": request.path_params.get("issueTypeId"),
        "propertyKey": request.path_params.get("propertyKey"),
        "headers": request.headers,
        "url": request.url,
    })

class DummyJiraClient:
"""A dummy JiraClient that returns a DummyHTTPClient."""
def init(self, base_url="https://dummy.atlassian.net"):
self._client = DummyHTTPClient(base_url=base_url)

def get_client(self):
    return self._client

--- Copy of the function under test and helpers (EXACT as provided) ---

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.status_code = response.status_code
self.data = response.data
from app.sources.external.jira.jira import JiraDataSource

--- TEST SUITE ---

1. Basic Test Cases

@pytest.mark.asyncio
async def test_get_issue_type_property_basic_success():
"""Test basic successful call returns expected HTTPResponse."""
client = DummyJiraClient()
ds = JiraDataSource(client)
resp = await ds.get_issue_type_property("10001", "myProp")

@pytest.mark.asyncio
async def test_get_issue_type_property_with_headers():
"""Test passing custom headers is reflected in the request."""
client = DummyJiraClient()
ds = JiraDataSource(client)
custom_headers = {"X-Test-Header": "test123"}
resp = await ds.get_issue_type_property("10002", "customKey", headers=custom_headers)

@pytest.mark.asyncio
async def test_get_issue_type_property_empty_headers():
"""Test that passing no headers does not cause errors."""
client = DummyJiraClient()
ds = JiraDataSource(client)
resp = await ds.get_issue_type_property("10003", "emptyHeader")

@pytest.mark.asyncio
async def test_get_issue_type_property_url_formatting():
"""Test that the constructed URL is correct."""
client = DummyJiraClient(base_url="https://myjira.net/")
ds = JiraDataSource(client)
resp = await ds.get_issue_type_property("abc", "xyz")
expected_url = "https://myjira.net/rest/api/3/issuetype/abc/properties/xyz"

2. Edge Test Cases

@pytest.mark.asyncio
async def test_get_issue_type_property_issue_type_not_found():
"""Test when the issueTypeId does not exist (simulate 404)."""
client = DummyJiraClient()
ds = JiraDataSource(client)
resp = await ds.get_issue_type_property("notfound", "anyKey")

@pytest.mark.asyncio

async def test_get_issue_type_property_none_client_raises():
"""Test that initialization with client returning None raises ValueError."""
class BadJiraClient:
def get_client(self):
return None
with pytest.raises(ValueError, match="HTTP client is not initialized"):
JiraDataSource(BadJiraClient())

@pytest.mark.asyncio
async def test_get_issue_type_property_client_missing_base_url():
"""Test that missing get_base_url method raises ValueError."""
class BadHTTPClient:
pass
class BadJiraClient:
def get_client(self):
return BadHTTPClient()
with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
JiraDataSource(BadJiraClient())

@pytest.mark.asyncio
async def test_get_issue_type_property_concurrent_calls():
"""Test concurrent execution with different parameters."""
client = DummyJiraClient()
ds = JiraDataSource(client)
params = [
("A", "prop1"),
("B", "prop2"),
("C", "prop3"),
("notfound", "prop4"),
]
# Run all requests concurrently
results = await asyncio.gather(
*(ds.get_issue_type_property(issueTypeId, propertyKey) for issueTypeId, propertyKey in params)
)

@pytest.mark.asyncio
async def test_get_issue_type_property_path_param_types():
"""Test that non-string types for path params are handled as strings."""
client = DummyJiraClient()
ds = JiraDataSource(client)
resp = await ds.get_issue_type_property(12345, 67890)

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_get_issue_type_property_many_concurrent_calls():
"""Test many concurrent calls for scalability."""
client = DummyJiraClient()
ds = JiraDataSource(client)
ids = [f"id{i}" for i in range(50)]
keys = [f"key{i}" for i in range(50)]
tasks = [
ds.get_issue_type_property(issueTypeId, propertyKey)
for issueTypeId, propertyKey in zip(ids, keys)
]
results = await asyncio.gather(*tasks)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_get_issue_type_property_concurrent_with_mixed_results():
"""Test concurrent calls with some errors and some successes."""
client = DummyJiraClient()
ds = JiraDataSource(client)
params = [
("id1", "key1"),
("notfound", "key2"),
("id3", "error"), # Will raise
("id4", "key4"),
]
# Use asyncio.gather with return_exceptions=True to capture exceptions
results = await asyncio.gather(
*(ds.get_issue_type_property(issueTypeId, propertyKey) for issueTypeId, propertyKey in params),
return_exceptions=True
)

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_get_issue_type_property_throughput_small_load():
"""Throughput: Test a small burst of concurrent requests."""
client = DummyJiraClient()
ds = JiraDataSource(client)
tasks = [ds.get_issue_type_property(f"id{i}", f"key{i}") for i in range(10)]
results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_get_issue_type_property_throughput_medium_load():
"""Throughput: Test a medium-sized burst of concurrent requests."""
client = DummyJiraClient()
ds = JiraDataSource(client)
tasks = [ds.get_issue_type_property(f"id{i}", f"key{i}") for i in range(100)]
results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_get_issue_type_property_throughput_mixed_load():
"""Throughput: Test a mix of valid and error-inducing requests."""
client = DummyJiraClient()
ds = JiraDataSource(client)
tasks = []
for i in range(40):
if i % 10 == 0:
tasks.append(ds.get_issue_type_property("notfound", f"key{i}"))
elif i % 15 == 0:
tasks.append(ds.get_issue_type_property(f"id{i}", "error"))
else:
tasks.append(ds.get_issue_type_property(f"id{i}", f"key{i}"))
results = await asyncio.gather(*tasks, return_exceptions=True)
# Check that errors and successes are as expected
notfound_count = sum(isinstance(r, DummyHTTPResponse) and r.status_code == 404 for r in results)
error_count = sum(isinstance(r, RuntimeError) for r in results)
ok_count = sum(isinstance(r, DummyHTTPResponse) and r.status_code == 200 for r in results)

@pytest.mark.asyncio
async def test_get_issue_type_property_throughput_high_volume():
"""Throughput: Test high volume concurrent requests (performance/scalability)."""
client = DummyJiraClient()
ds = JiraDataSource(client)
# Keep under 1000 as per instructions, using 200 for speed
tasks = [ds.get_issue_type_property(f"id{i}", f"key{i}") for i in range(200)]
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

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

--- Minimal stubs for HTTPRequest and HTTPResponse ---

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

--- Minimal stub for JiraClient and HTTP client ---

class DummyHTTPClient:
"""A dummy async HTTP client that simulates HTTP request execution."""

def __init__(self, base_url, execute_behavior=None):
    self._base_url = base_url
    self._execute_behavior = execute_behavior or (lambda req: HTTPResponse({"url": req.url, "headers": req.headers, "path_params": req.path_params, "query_params": req.query_params, "body": req.body}))

def get_base_url(self):
    return self._base_url

async def execute(self, req):
    # Simulate async HTTP call
    return self._execute_behavior(req)

class DummyJiraClient:
"""Simulates the JiraClient interface."""

def __init__(self, http_client):
    self._client = http_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_issue_type_property_basic_success():
"""Test basic async/await behavior and correct URL construction."""
base_url = "https://jira.example.com"
dummy_client = DummyHTTPClient(base_url)
jira_client = DummyJiraClient(dummy_client)
ds = JiraDataSource(jira_client)

# Await the async function and check response
resp = await ds.get_issue_type_property("123", "mykey")

@pytest.mark.asyncio
async def test_get_issue_type_property_with_headers():
"""Test passing custom headers."""
base_url = "https://jira.example.com"
dummy_client = DummyHTTPClient(base_url)
jira_client = DummyJiraClient(dummy_client)
ds = JiraDataSource(jira_client)

resp = await ds.get_issue_type_property("456", "otherkey", headers={"X-Test": "abc", "Content-Type": "application/json"})

@pytest.mark.asyncio
async def test_get_issue_type_property_empty_headers():
"""Test with headers=None and headers={}."""
base_url = "https://jira.example.com"
dummy_client = DummyHTTPClient(base_url)
jira_client = DummyJiraClient(dummy_client)
ds = JiraDataSource(jira_client)

resp1 = await ds.get_issue_type_property("789", "empty", headers=None)
resp2 = await ds.get_issue_type_property("789", "empty", headers={})

2. Edge Test Cases

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

@pytest.mark.asyncio
async def test_get_issue_type_property_missing_get_base_url():
"""Test ValueError when HTTP client lacks get_base_url method."""
class NoBaseUrlClient:
pass
class NoBaseUrlJiraClient:
def get_client(self):
return NoBaseUrlClient()
with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
JiraDataSource(NoBaseUrlJiraClient())

@pytest.mark.asyncio
async def test_get_issue_type_property_concurrent_calls():
"""Test concurrent execution of multiple calls with different arguments."""
base_url = "https://jira.example.com"
dummy_client = DummyHTTPClient(base_url)
jira_client = DummyJiraClient(dummy_client)
ds = JiraDataSource(jira_client)

async def call(idx):
    return await ds.get_issue_type_property(str(idx), f"key{idx}")

results = await asyncio.gather(*(call(i) for i in range(10)))
for i, resp in enumerate(results):
    pass

@pytest.mark.asyncio
async def test_get_issue_type_property_execute_raises():
"""Test exception handling when the underlying client raises an exception."""
class FailingHTTPClient(DummyHTTPClient):
async def execute(self, req):
raise RuntimeError("Simulated HTTP failure")
jira_client = DummyJiraClient(FailingHTTPClient("https://jira.example.com"))
ds = JiraDataSource(jira_client)
with pytest.raises(RuntimeError, match="Simulated HTTP failure"):
await ds.get_issue_type_property("fail", "failkey")

@pytest.mark.asyncio
async def test_get_issue_type_property_special_characters():
"""Test with special characters in issueTypeId and propertyKey."""
base_url = "https://jira.example.com"
dummy_client = DummyHTTPClient(base_url)
jira_client = DummyJiraClient(dummy_client)
ds = JiraDataSource(jira_client)
issueTypeId = "id/with/slash"
propertyKey = "prop?key=val"
resp = await ds.get_issue_type_property(issueTypeId, propertyKey)

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_get_issue_type_property_many_concurrent_calls():
"""Test a large number of concurrent calls (up to 100)."""
base_url = "https://jira.example.com"
dummy_client = DummyHTTPClient(base_url)
jira_client = DummyJiraClient(dummy_client)
ds = JiraDataSource(jira_client)

async def call(idx):
    return await ds.get_issue_type_property(f"id{idx}", f"key{idx}")

tasks = [call(i) for i in range(100)]
results = await asyncio.gather(*tasks)
for i, resp in enumerate(results):
    pass

@pytest.mark.asyncio
async def test_get_issue_type_property_large_headers():
"""Test with a large headers dict."""
base_url = "https://jira.example.com"
dummy_client = DummyHTTPClient(base_url)
jira_client = DummyJiraClient(dummy_client)
ds = JiraDataSource(jira_client)
headers = {f"X-Header-{i}": f"val{i}" for i in range(200)}
resp = await ds.get_issue_type_property("big", "header", headers=headers)
for i in range(200):
pass

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_get_issue_type_property_throughput_small_load():
"""Throughput test: small load (10 calls)."""
base_url = "https://jira.example.com"
dummy_client = DummyHTTPClient(base_url)
jira_client = DummyJiraClient(dummy_client)
ds = JiraDataSource(jira_client)

async def call(idx):
    return await ds.get_issue_type_property(f"sm{idx}", f"key{idx}")

results = await asyncio.gather(*(call(i) for i in range(10)))
for i, resp in enumerate(results):
    pass

@pytest.mark.asyncio
async def test_get_issue_type_property_throughput_medium_load():
"""Throughput test: medium load (50 calls)."""
base_url = "https://jira.example.com"
dummy_client = DummyHTTPClient(base_url)
jira_client = DummyJiraClient(dummy_client)
ds = JiraDataSource(jira_client)

async def call(idx):
    return await ds.get_issue_type_property(f"md{idx}", f"key{idx}")

results = await asyncio.gather(*(call(i) for i in range(50)))
for i, resp in enumerate(results):
    pass

@pytest.mark.asyncio
async def test_get_issue_type_property_throughput_high_volume():
"""Throughput test: high volume (200 calls)."""
base_url = "https://jira.example.com"
dummy_client = DummyHTTPClient(base_url)
jira_client = DummyJiraClient(dummy_client)
ds = JiraDataSource(jira_client)

async def call(idx):
    return await ds.get_issue_type_property(f"hv{idx}", f"key{idx}")

results = await asyncio.gather(*(call(i) for i in range(200)))
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_property-mhs49bjd and push.