⚡️ Speed up method JiraDataSource.create_issue_type by 11%

[codeflash-ai]

📄 11% (0.11x) speedup for JiraDataSource.create_issue_type in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 2.34 milliseconds → 2.11 milliseconds (best of 250 runs)

📝 Explanation and details

The optimized code achieves a 10% runtime improvement through several key micro-optimizations in the create_issue_type method:

Key Optimizations Applied:

1. Reduced Dictionary Operations: The original code created three unnecessary empty dictionaries (_path, _query, and their corresponding _as_str_dict calls). The optimized version eliminates these by directly passing empty dictionaries {} to the HTTPRequest constructor, saving ~39% of _as_str_dict processing time (from 1710 hits to 570 hits in line profiler).

2. Optimized Client Access: Cached self._client in a local variable client to reduce attribute lookups during the async execution path.

3. Streamlined Header Processing: Moved header dictionary creation (dict(headers or ())) closer to usage and used an empty tuple as default instead of empty dict, reducing object creation overhead.

4. Simplified URL Construction: Pass empty dict {} directly to _safe_format_url instead of creating and passing the unused _path variable.

Performance Impact Analysis:

The line profiler shows the most significant gains in _as_str_dict function calls, which dropped from 1.91ms total time to 1.16ms (39% reduction). This function was called 3x per request in the original (for headers, path_params, query_params) but only 1x in the optimized version (just headers).

Test Case Performance:

Based on the annotated tests, these optimizations are particularly beneficial for:

• High-throughput scenarios (medium/large load tests with 50-200+ concurrent requests)
• Batch operations where the function is called repeatedly
• Resource-constrained environments where minimizing object allocation matters

The optimizations maintain identical functionality and async behavior while reducing computational overhead, making them especially valuable for API clients that may be called frequently in production workloads.

✅ Correctness verification report:

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

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

--- Fake HTTP Client for testing ---

class FakeAsyncClient:
def init(self):
self.executed_requests = []

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

async def execute(self, req):
    self.executed_requests.append(req)
    # Simulate a response based on the request for testing
    # Return the body for easy verification
    return HTTPResponse({
        "url": req.url,
        "method": req.method,
        "headers": req.headers,
        "body": req.body,
        "path_params": req.path_params,
        "query_params": req.query_params
    })

class FakeJiraClient:
def init(self, client=None):
self._client = client if client is not None else FakeAsyncClient()
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_create_issue_type_basic_required_fields():
"""Test basic creation with only required 'name' field."""
ds = JiraDataSource(FakeJiraClient())
resp = await ds.create_issue_type(name="Bug")

@pytest.mark.asyncio
async def test_create_issue_type_all_fields():
"""Test creation with all fields provided."""
ds = JiraDataSource(FakeJiraClient())
resp = await ds.create_issue_type(
name="Task",
description="A task issue type",
hierarchyLevel=2,
type="standard",
headers={"X-Custom": "value"}
)

@pytest.mark.asyncio
async def test_create_issue_type_custom_headers_override():
"""Test that a custom Content-Type header is respected."""
ds = JiraDataSource(FakeJiraClient())
resp = await ds.create_issue_type(
name="Epic",
headers={"Content-Type": "application/x-custom"}
)

2. Edge Test Cases

@pytest.mark.asyncio
async def test_create_issue_type_empty_name():
"""Test with empty string for name (still required)."""
ds = JiraDataSource(FakeJiraClient())
resp = await ds.create_issue_type(name="")

@pytest.mark.asyncio
async def test_create_issue_type_none_optional_fields():
"""Test with all optional fields as None."""
ds = JiraDataSource(FakeJiraClient())
resp = await ds.create_issue_type(
name="Story",
description=None,
hierarchyLevel=None,
type=None,
headers=None
)

@pytest.mark.asyncio
async def test_create_issue_type_concurrent_execution():
"""Test concurrent execution of multiple create_issue_type calls."""
ds = JiraDataSource(FakeJiraClient())
# Launch 5 concurrent calls with different names
names = [f"IssueType_{i}" for i in range(5)]
coros = [ds.create_issue_type(name=n) for n in names]
results = await asyncio.gather(*coros)
# Each result should have the corresponding name
for i, resp in enumerate(results):
pass

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

@pytest.mark.asyncio
async def test_create_issue_type_client_missing_get_base_url():
"""Test error if client does not have get_base_url method."""
class NoBaseUrlClient:
def get_client(self):
return object()
with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
JiraDataSource(NoBaseUrlClient())

@pytest.mark.asyncio
async def test_create_issue_type_headers_are_stringified():
"""Test that non-string header keys/values are stringified."""
ds = JiraDataSource(FakeJiraClient())
resp = await ds.create_issue_type(
name="HeaderTest",
headers={123: 456}
)

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_create_issue_type_large_batch_concurrent():
"""Test concurrent creation of many issue types (up to 50)."""
ds = JiraDataSource(FakeJiraClient())
names = [f"Bulk_{i}" for i in range(50)]
coros = [ds.create_issue_type(name=n, hierarchyLevel=i) for i, n in enumerate(names)]
results = await asyncio.gather(*coros)
# All responses should be correct and unique
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_create_issue_type_long_description():
"""Test with a very long description string."""
ds = JiraDataSource(FakeJiraClient())
long_desc = "x" * 4096 # 4KB description
resp = await ds.create_issue_type(name="LongDesc", description=long_desc)

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_create_issue_type_throughput_small_load():
"""Throughput test: small batch of 10 concurrent requests."""
ds = JiraDataSource(FakeJiraClient())
coros = [ds.create_issue_type(name=f"Small_{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_type_throughput_medium_load():
"""Throughput test: medium batch of 100 concurrent requests."""
ds = JiraDataSource(FakeJiraClient())
coros = [ds.create_issue_type(name=f"Med_{i}", hierarchyLevel=i%5) 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_type_throughput_mixed_fields():
"""Throughput test: mixed field values in concurrent requests."""
ds = JiraDataSource(FakeJiraClient())
coros = [
ds.create_issue_type(
name=f"Mix_{i}",
description=None if i % 2 == 0 else f"desc_{i}",
hierarchyLevel=i if i % 3 == 0 else None,
type="subtask" if i % 4 == 0 else None
)
for i in range(20)
]
results = await asyncio.gather(*coros)
for i, resp in enumerate(results):
if i % 2 == 0:
pass
else:
pass
if i % 3 == 0:
pass
else:
pass
if i % 4 == 0:
pass
else:
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

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

---- Minimal stubs for dependencies ----

These are minimal implementations to allow the tests to run without external dependencies.

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

def json(self):
    # Return the response data as a dict
    return self.response_data

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 DummyHTTPClient:
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):
    # Simulate a response containing the request's body and headers
    self.executed_requests.append(request)
    return HTTPResponse({
        "url": request.url,
        "method": request.method,
        "headers": request.headers,
        "body": request.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_create_issue_type_basic_minimal():
"""Test basic creation with only required 'name' parameter."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
response = await ds.create_issue_type(name="Bug")
# Check that description, hierarchyLevel, and type are not present
body = response.json()["body"]

@pytest.mark.asyncio
async def test_create_issue_type_basic_all_fields():
"""Test creation with all optional fields provided."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
response = await ds.create_issue_type(
name="Story",
description="A user story",
hierarchyLevel=1,
type="standard"
)
body = response.json()["body"]

@pytest.mark.asyncio
async def test_create_issue_type_basic_custom_headers():
"""Test that custom headers are merged and Content-Type is set."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
custom_headers = {"X-Custom-Header": "value"}
response = await ds.create_issue_type(
name="Task",
headers=custom_headers
)
headers = response.json()["headers"]

2. Edge Test Cases

@pytest.mark.asyncio
async def test_create_issue_type_edge_empty_name():
"""Test edge case where name is empty string."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
response = await ds.create_issue_type(name="")

@pytest.mark.asyncio
async def test_create_issue_type_edge_none_optional_fields():
"""Test edge case where all optional fields are None."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
response = await ds.create_issue_type(
name="Epic",
description=None,
hierarchyLevel=None,
type=None,
headers=None
)
body = response.json()["body"]

@pytest.mark.asyncio
async def test_create_issue_type_edge_long_name_and_description():
"""Test edge case with very long name and description."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
long_name = "X" * 512
long_description = "Y" * 2048
response = await ds.create_issue_type(
name=long_name,
description=long_description
)
body = response.json()["body"]

@pytest.mark.asyncio
async def test_create_issue_type_edge_invalid_hierarchy_level():
"""Test edge case with negative hierarchyLevel."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
response = await ds.create_issue_type(
name="Subtask",
hierarchyLevel=-1
)
body = response.json()["body"]

@pytest.mark.asyncio
async def test_create_issue_type_edge_concurrent_execution():
"""Test concurrent execution of multiple create_issue_type calls."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
names = [f"IssueType{i}" for i in range(10)]
coros = [ds.create_issue_type(name=n) for n in names]
responses = await asyncio.gather(*coros)
# Check that each response contains the correct name
for i, resp in enumerate(responses):
pass

@pytest.mark.asyncio
async def test_create_issue_type_edge_missing_client():
"""Test edge case where the HTTP client is not initialized."""
class DummyJiraClient:
def get_client(self):
return None
with pytest.raises(ValueError) as excinfo:
JiraDataSource(DummyJiraClient())

@pytest.mark.asyncio
async def test_create_issue_type_edge_client_missing_get_base_url():
"""Test edge case where client does not have get_base_url method."""
class BadClient:
pass
class DummyJiraClient:
def get_client(self):
return BadClient()
with pytest.raises(ValueError) as excinfo:
JiraDataSource(DummyJiraClient())

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_create_issue_type_large_scale_many_concurrent():
"""Test large scale with many concurrent create_issue_type calls."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
num_requests = 100
names = [f"Type_{i}" for i in range(num_requests)]
coros = [ds.create_issue_type(name=n, description=f"Desc_{i}") for i, n in enumerate(names)]
responses = await asyncio.gather(*coros)
for i, resp in enumerate(responses):
body = resp.json()["body"]

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_create_issue_type_throughput_small_load():
"""Throughput test: small load (10 requests)."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
coros = [ds.create_issue_type(name=f"Small_{i}") for i in range(10)]
responses = await asyncio.gather(*coros)
for i, resp in enumerate(responses):
pass

@pytest.mark.asyncio
async def test_create_issue_type_throughput_medium_load():
"""Throughput test: medium load (50 requests)."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
coros = [ds.create_issue_type(name=f"Medium_{i}", type="standard") for i in range(50)]
responses = await asyncio.gather(*coros)
for i, resp in enumerate(responses):
pass

@pytest.mark.asyncio
async def test_create_issue_type_throughput_large_load():
"""Throughput test: large load (200 requests)."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
coros = [ds.create_issue_type(name=f"Large_{i}", hierarchyLevel=i % 5) for i in range(200)]
responses = await asyncio.gather(*coros)
for i, resp in enumerate(responses):
pass

5. Async context manager/iterator edge case (not directly applicable, but test for proper async/await usage)

@pytest.mark.asyncio
async def test_create_issue_type_async_await_usage():
"""Test that the function returns a coroutine and must be awaited."""
client = DummyHTTPClient("https://jira.example.com/")
ds = JiraDataSource(JiraClient(client))
codeflash_output = ds.create_issue_type(name="AwaitTest"); coro = codeflash_output
resp = await coro

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_type-mhrz3hds and push.