⚡️ Speed up method JiraDataSource.create_issue_security_scheme by 28%

[codeflash-ai]

📄 28% (0.28x) speedup for JiraDataSource.create_issue_security_scheme in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 3.27 milliseconds → 2.55 milliseconds (best of 250 runs)

📝 Explanation and details

The optimized code achieves a 28% runtime improvement (3.27ms → 2.55ms) and 3.7% throughput increase through several targeted micro-optimizations that reduce function call overhead and dictionary allocations:

Key Performance Optimizations:

1. Smarter header handling: Replaced dict(headers or {}) with headers.copy() if headers else {}, avoiding unnecessary dict construction when headers is None. Added explicit 'Content-Type' not in _headers check instead of using setdefault(), reducing redundant work when the header already exists.

2. Eliminated redundant parameter checking: Removed the expensive 'body_additional' in locals() check (which requires local variable lookup) and replaced it with the simpler if body_additional: test, since the parameter is always available.

3. Fast-path optimizations in helper functions:

   • _safe_format_url: Added early return for static endpoints (if not params or '{' not in template), avoiding expensive format_map() calls and exception handling for the common case where no URL formatting is needed.
   • _as_str_dict: Added if not d: return {} fast-path for empty dictionaries, eliminating unnecessary loops and function calls.

4. Inlined _serialize_value function: Moved the serialization logic directly into the file to reduce import overhead and function call costs, which is significant since _as_str_dict appears in profiling hotspots.

5. Optimized HTTPClient header merging: Replaced unconditional dict unpacking ({**self.headers, **request.headers}) with conditional logic that only copies headers when needed, reducing allocations.

Profile Impact Analysis:
The line profiler shows the most significant improvements in _safe_format_url (806μs → 265μs) due to the early return optimization, and modest improvements in _as_str_dict despite the function being called 2061 times across all requests.

Test Performance Context:
Based on the annotated tests, these optimizations are particularly effective for:

• High-throughput scenarios: Concurrent execution tests (50+ requests) benefit from reduced per-request overhead
• Repeated API calls: The helper function optimizations compound when processing many requests
• Basic usage patterns: Even simple cases with minimal parameters see improvements from reduced dictionary allocations

The optimizations maintain full compatibility while providing meaningful performance gains for both single requests and batch processing scenarios typical in Jira API integration workflows.

✅ Correctness verification report:

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

🌀 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 HTTPRequest/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, response_data):
self.data = response_data
def json(self):
return self.data

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

class DummyAsyncClient:
def init(self):
self.requests = []
async def request(self, method, url, **kwargs):
# Simulate a response object
self.requests.append((method, url, kwargs))
# Return a dict with all request info for validation
return {
"method": method,
"url": url,
"headers": kwargs.get("headers", {}),
"params": kwargs.get("params", {}),
"json": kwargs.get("json", {}),
"data": kwargs.get("data", {}),
"content": kwargs.get("content", None),
}

class DummyHTTPClient:
def init(self, base_url="http://localhost", token="dummy"):
self.headers = {"Authorization": f"Bearer {token}"}
self.base_url = base_url
self.client = DummyAsyncClient()
def get_base_url(self):
return self.base_url
async def execute(self, request, **kwargs):
# Simulate an HTTP request
response = await self.client.request(request.method, request.url, headers=request.headers, params=request.query_params, json=request.body)
return HTTPResponse(response)
async def _ensure_client(self):
return self.client

class DummyJiraClient:
def init(self, client):
self.client = client
def get_client(self):
return self.client
from app.sources.external.jira.jira import JiraDataSource

---- TEST CASES ----

--- Basic Test Cases ---

@pytest.mark.asyncio
async def test_create_issue_security_scheme_basic_required_only():
"""Test with only required parameter 'name'."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
resp = await ds.create_issue_security_scheme(name="TestScheme")

@pytest.mark.asyncio
async def test_create_issue_security_scheme_basic_all_fields():
"""Test with all parameters provided."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
levels = [{"id": 1, "name": "Level1"}, {"id": 2, "name": "Level2"}]
resp = await ds.create_issue_security_scheme(
name="SchemeFull",
description="A test scheme",
levels=levels,
body_additional={"extra": "value"},
headers={"X-Test-Header": "abc"}
)

@pytest.mark.asyncio
async def test_create_issue_security_scheme_basic_body_additional_merges():
"""Test that body_additional merges correctly with other fields."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
resp = await ds.create_issue_security_scheme(
name="SchemeWithExtra",
body_additional={"description": "Overwritten", "custom": 123}
)

--- Edge Test Cases ---

@pytest.mark.asyncio
async def test_create_issue_security_scheme_edge_empty_levels():
"""Test with empty levels list."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
resp = await ds.create_issue_security_scheme(
name="EmptyLevelsScheme",
levels=[]
)

@pytest.mark.asyncio
async def test_create_issue_security_scheme_edge_none_body_additional():
"""Test with body_additional=None."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
resp = await ds.create_issue_security_scheme(
name="NoneBodyAdditional",
body_additional=None
)

@pytest.mark.asyncio
async def test_create_issue_security_scheme_edge_headers_override_content_type():
"""Test headers override default content-type."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
resp = await ds.create_issue_security_scheme(
name="OverrideHeader",
headers={"Content-Type": "application/x-www-form-urlencoded"}
)

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

@pytest.mark.asyncio
async def test_create_issue_security_scheme_edge_missing_base_url_method_raises():
"""Test that missing get_base_url 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_create_issue_security_scheme_concurrent_execution():
"""Test concurrent execution of multiple requests."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
# Run 5 requests concurrently
names = [f"Scheme{i}" for i in range(5)]
coros = [ds.create_issue_security_scheme(name=n) for n in names]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
pass

--- Large Scale Test Cases ---

@pytest.mark.asyncio
async def test_create_issue_security_scheme_large_scale_many_concurrent():
"""Test many concurrent requests (50)."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
coros = [
ds.create_issue_security_scheme(
name=f"BulkScheme{i}",
description=f"Desc{i}",
levels=[{"id": i, "name": f"Level{i}"}]
)
for i in range(50)
]
results = await asyncio.gather(*coros)
# Check all results
for i, resp in enumerate(results):
pass

--- Throughput Test Cases ---

@pytest.mark.asyncio
async def test_create_issue_security_scheme_throughput_small_load():
"""Throughput test with small load (10 requests)."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
coros = [ds.create_issue_security_scheme(name=f"SmallLoad{i}") for i in range(10)]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_create_issue_security_scheme_throughput_medium_load():
"""Throughput test with medium load (100 requests)."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
coros = [ds.create_issue_security_scheme(name=f"MediumLoad{i}") for i in range(100)]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_create_issue_security_scheme_throughput_large_load():
"""Throughput test with large load (250 requests)."""
client = DummyJiraClient(DummyHTTPClient())
ds = JiraDataSource(client)
coros = [ds.create_issue_security_scheme(name=f"LargeLoad{i}") for i in range(250)]
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.

#------------------------------------------------
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 isolated testing ----

class DummyHTTPResponse:
"""A dummy HTTPResponse object for testing."""
def init(self, status_code=200, json_data=None, text_data=None, request=None):
self.status_code = status_code
self._json_data = json_data or {}
self.text = text_data or ""
self.request = request

def json(self):
    return self._json_data

class DummyHTTPClient:
"""A dummy async HTTP client that records requests and returns a dummy response."""
def init(self, base_url="https://dummy.atlassian.net"):
self._base_url = base_url
self.requests = []
self.closed = False

def get_base_url(self):
    return self._base_url

async def execute(self, request):
    """Simulate async HTTP execution by recording the request and returning a dummy response."""
    self.requests.append(request)
    # Echo back request data in the response for validation
    return DummyHTTPResponse(
        status_code=201,
        json_data={
            "received": {
                "method": request.method,
                "url": request.url,
                "headers": request.headers,
                "path_params": request.path_params,
                "query_params": request.query_params,
                "body": request.body,
            }
        },
        request=request
    )

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-paste of the function under test and helpers (DO NOT MODIFY) ----

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
from app.sources.external.jira.jira import JiraDataSource

---- TESTS BELOW ----

1. Basic Test Cases

@pytest.mark.asyncio
async def test_create_issue_security_scheme_basic_minimal():
"""Test basic creation with only required 'name' argument."""
ds = JiraDataSource(DummyJiraClient())
response = await ds.create_issue_security_scheme(name="TestScheme")
# The body must contain the name only
body = response.json()["received"]["body"]

@pytest.mark.asyncio
async def test_create_issue_security_scheme_basic_full_args():
"""Test creation with all arguments provided."""
ds = JiraDataSource(DummyJiraClient())
levels = [{"id": 1, "name": "Level1"}, {"id": 2, "name": "Level2"}]
body_additional = {"customField": "customValue"}
headers = {"Authorization": "Bearer token123"}
response = await ds.create_issue_security_scheme(
name="FullScheme",
description="A full test scheme",
levels=levels,
body_additional=body_additional,
headers=headers
)
# Check that all fields are present in the request body
body = response.json()["received"]["body"]
# Check that headers are merged and Content-Type is set
sent_headers = response.json()["received"]["headers"]

@pytest.mark.asyncio
async def test_create_issue_security_scheme_basic_custom_headers_override():
"""Test that custom Content-Type header is respected and not overwritten."""
ds = JiraDataSource(DummyJiraClient())
headers = {"Content-Type": "application/x-custom"}
response = await ds.create_issue_security_scheme(name="HeaderTest", headers=headers)
sent_headers = response.json()["received"]["headers"]

2. Edge Test Cases

@pytest.mark.asyncio
async def test_create_issue_security_scheme_edge_empty_levels_and_body_additional():
"""Test with empty levels list and empty body_additional dict."""
ds = JiraDataSource(DummyJiraClient())
response = await ds.create_issue_security_scheme(
name="EdgeCase",
levels=[],
body_additional={}
)
# Should include name and empty levels
body = response.json()["received"]["body"]

@pytest.mark.asyncio
async def test_create_issue_security_scheme_edge_none_headers_and_body_additional():
"""Test with headers=None and body_additional=None."""
ds = JiraDataSource(DummyJiraClient())
response = await ds.create_issue_security_scheme(
name="NoneCase",
headers=None,
body_additional=None
)
body = response.json()["received"]["body"]
headers = response.json()["received"]["headers"]

@pytest.mark.asyncio
async def test_create_issue_security_scheme_edge_invalid_client_raises():
"""Test that ValueError is raised if client.get_client() returns 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_create_issue_security_scheme_edge_client_missing_get_base_url():
"""Test that ValueError is raised if client does not have get_base_url()."""
class NoBaseUrlClient:
def get_client(self):
class Dummy:
pass
return Dummy()
with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
JiraDataSource(NoBaseUrlClient())

@pytest.mark.asyncio
async def test_create_issue_security_scheme_edge_concurrent_execution():
"""Test concurrent execution of multiple create_issue_security_scheme calls."""
ds = JiraDataSource(DummyJiraClient())
# Prepare several different calls
names = [f"Scheme{i}" for i in range(10)]
coros = [ds.create_issue_security_scheme(name=n) for n in names]
responses = await asyncio.gather(*coros)
# All responses must be correct and unique
for i, resp in enumerate(responses):
body = resp.json()["received"]["body"]

@pytest.mark.asyncio
async def test_create_issue_security_scheme_edge_body_additional_overwrites():
"""Test that body_additional can overwrite other body fields."""
ds = JiraDataSource(DummyJiraClient())
body_additional = {"name": "Overwritten", "description": "OverwrittenDesc"}
response = await ds.create_issue_security_scheme(
name="OriginalName",
description="OriginalDesc",
body_additional=body_additional
)
# body_additional should overwrite name and description
body = response.json()["received"]["body"]

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_create_issue_security_scheme_large_scale_many_concurrent():
"""Test large number of concurrent create_issue_security_scheme calls."""
ds = JiraDataSource(DummyJiraClient())
N = 50 # Keep under 1000 as per instructions
coros = [
ds.create_issue_security_scheme(
name=f"BulkScheme{i}",
description=f"BulkDesc{i}",
levels=[{"id": i, "name": f"Level{i}"}],
body_additional={"extra": i}
)
for i in range(N)
]
responses = await asyncio.gather(*coros)
# Validate all responses
for i, resp in enumerate(responses):
body = resp.json()["received"]["body"]

@pytest.mark.asyncio
async def test_create_issue_security_scheme_large_scale_unique_headers():
"""Test concurrent calls with unique headers for each request."""
ds = JiraDataSource(DummyJiraClient())
N = 20
coros = [
ds.create_issue_security_scheme(
name=f"HeaderScheme{i}",
headers={"X-Request-ID": f"req-{i}"}
)
for i in range(N)
]
responses = await asyncio.gather(*coros)
for i, resp in enumerate(responses):
headers = resp.json()["received"]["headers"]

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_create_issue_security_scheme_throughput_small_load():
"""Throughput: Test performance under small concurrent load."""
ds = JiraDataSource(DummyJiraClient())
coros = [ds.create_issue_security_scheme(name=f"SmallLoad{i}") for i in range(5)]
responses = await asyncio.gather(*coros)
# Ensure all names are correct
for i, r in enumerate(responses):
pass

@pytest.mark.asyncio
async def test_create_issue_security_scheme_throughput_medium_load():
"""Throughput: Test performance under medium concurrent load."""
ds = JiraDataSource(DummyJiraClient())
coros = [ds.create_issue_security_scheme(name=f"MediumLoad{i}") for i in range(25)]
responses = await asyncio.gather(*coros)
# Ensure all names are correct
for i, r in enumerate(responses):
pass

@pytest.mark.asyncio
async def test_create_issue_security_scheme_throughput_large_load():
"""Throughput: Test performance under large but bounded concurrent load."""
ds = JiraDataSource(DummyJiraClient())
N = 100 # Stay well below 1000 for test speed
coros = [ds.create_issue_security_scheme(name=f"LargeLoad{i}") for i in range(N)]
responses = await asyncio.gather(*coros)
# Ensure all names are correct
for i, r in enumerate(responses):
pass

@pytest.mark.asyncio
async def test_create_issue_security_scheme_throughput_sustained_pattern():
"""Throughput: Test sustained execution pattern in batches."""
ds = JiraDataSource(DummyJiraClient())
batch_size = 10
batches = 5
all_responses = []
for batch in range(batches):
coros = [
ds.create_issue_security_scheme(name=f"Sustained{batch}_{i}")
for i in range(batch_size)
]
responses = await asyncio.gather(*coros)
all_responses.extend(responses)
for resp in all_responses:
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.create_issue_security_scheme-mhrrzvih and push.