⚡️ Speed up method JiraDataSource.assign_issue_type_screen_scheme_to_project by 9%

[codeflash-ai]

📄 9% (0.09x) speedup for JiraDataSource.assign_issue_type_screen_scheme_to_project in backend/python/app/sources/external/jira/jira.py

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

📝 Explanation and details

The optimized code achieves an 8% runtime improvement and 2.9% throughput increase through several targeted micro-optimizations:

Key Optimizations:

1. Eliminated unnecessary dict allocations: Removed creation of empty _path and _query dictionaries, passing empty dict literals {} directly to function calls. This saves memory allocation overhead on each function call.

2. Optimized dict creation for headers: Changed dict(headers or {}) to dict(headers or ()), using an empty tuple instead of empty dict when headers is None. This is slightly more efficient as tuples have less overhead than dicts for the falsy case.

3. Streamlined parameter passing: Instead of creating temporary _path and _query variables that were always empty, the code now passes empty dicts directly to HTTPRequest constructor and helper functions.

4. Improved header merging in HTTPClient: Replaced dictionary unpacking {**self.headers, **request.headers} with explicit copy and update operations (self.headers.copy() + update()), which is more memory-efficient for typical header sizes.

5. Conditional URL formatting: Added a conditional check if request.path_params else request.url to skip string formatting when path_params is empty, avoiding unnecessary format operations.

Performance Impact:
The line profiler shows the most significant gains in _as_str_dict function calls (from 1.93ms to 1.23ms total time) due to fewer dictionary operations. The assign_issue_type_screen_scheme_to_project function itself improved from 11.74ms to 9.60ms total time.

These optimizations are particularly effective for the test cases involving multiple concurrent requests, where the reduced per-call overhead compounds across many operations. The improvements benefit all API call patterns, from single requests to high-volume concurrent scenarios.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 591 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 stubs for HTTPRequest and HTTPResponse, as used by the function.

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

Minimal stub for JiraRESTClientViaApiKey

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

def get_base_url(self):
    return self.base_url

async def execute(self, request):
    # Simulate a successful HTTPResponse with request info for test verification
    return HTTPResponse({
        "method": request.method,
        "url": request.url,
        "headers": request.headers,
        "path_params": request.path_params,
        "query_params": request.query_params,
        "body": request.body,
    })

Minimal stub for JiraClient

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

Helper to create a valid JiraDataSource for tests

def make_datasource(base_url="https://jira.example.com"):
client = JiraRESTClientViaApiKey(base_url, "user@example.com", "api_key")
return JiraDataSource(JiraClient(client))

----------- 1. Basic Test Cases -----------

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_basic():
# Test with both parameters provided
ds = make_datasource()
resp = await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId="123",
projectId="456"
)
data = resp.response

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_only_issue_type():
# Test with only issueTypeScreenSchemeId provided
ds = make_datasource()
resp = await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId="abc"
)
data = resp.response

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_only_project():
# Test with only projectId provided
ds = make_datasource()
resp = await ds.assign_issue_type_screen_scheme_to_project(
projectId="xyz"
)
data = resp.response

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_no_params():
# Test with no parameters
ds = make_datasource()
resp = await ds.assign_issue_type_screen_scheme_to_project()
data = resp.response

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_custom_headers():
# Test with custom headers
ds = make_datasource()
resp = await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId="id",
projectId="pid",
headers={"Authorization": "Bearer token", "Content-Type": "application/json"}
)
data = resp.response

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

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_empty_strings():
# Test with empty string parameters
ds = make_datasource()
resp = await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId="",
projectId=""
)
data = resp.response

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_none_headers():
# Test with headers explicitly set to None
ds = make_datasource()
resp = await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId="id",
projectId="pid",
headers=None
)
data = resp.response

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_invalid_client():
# Test with None client, should raise ValueError
class DummyClient:
def get_client(self):
return None
with pytest.raises(ValueError, match="HTTP client is not initialized"):
JiraDataSource(DummyClient())

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_missing_base_url():
# Test with client missing get_base_url method, should raise ValueError
class DummyClient:
def get_client(self):
return object() # No get_base_url method
with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
JiraDataSource(DummyClient())

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_concurrent_execution():
# Test concurrent execution of the async function
ds = make_datasource()
async def call_func(i):
return await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId=f"id-{i}",
projectId=f"pid-{i}"
)
results = await asyncio.gather(*(call_func(i) for i in range(5)))
for i, resp in enumerate(results):
data = resp.response

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

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_many_concurrent():
# Test with 50 concurrent calls
ds = make_datasource()
async def call_func(i):
return await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId=f"scheme-{i}",
projectId=f"proj-{i}"
)
results = await asyncio.gather(*(call_func(i) for i in range(50)))
for i, resp in enumerate(results):
data = resp.response

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_large_body():
# Test with large strings for parameters
ds = make_datasource()
large_id = "x" * 500
large_project = "y" * 500
resp = await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId=large_id,
projectId=large_project
)
data = resp.response

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

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_throughput_small_load():
# Throughput test: small load (10 requests)
ds = make_datasource()
async def call_func(i):
return await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId=f"small-{i}",
projectId=f"smallproj-{i}"
)
results = await asyncio.gather(*(call_func(i) for i in range(10)))
for i, resp in enumerate(results):
data = resp.response

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_throughput_medium_load():
# Throughput test: medium load (100 requests)
ds = make_datasource()
async def call_func(i):
return await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId=f"medium-{i}",
projectId=f"mediumproj-{i}"
)
results = await asyncio.gather(*(call_func(i) for i in range(100)))
for i, resp in enumerate(results):
data = resp.response

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_throughput_high_volume():
# Throughput test: high volume (200 requests, but <1000 for performance)
ds = make_datasource()
async def call_func(i):
return await ds.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId=f"high-{i}",
projectId=f"highproj-{i}"
)
results = await asyncio.gather(*(call_func(i) for i in range(200)))
for i, resp in enumerate(results):
data = resp.response

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 and helpers to support the test environment ---

class HTTPResponse:
"""Minimal stub for HTTPResponse."""
def init(self, response_data):
self.response_data = response_data

def json(self):
    return self.response_data

class HTTPRequest:
"""Minimal stub for 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 DummyAsyncClient:
"""A dummy async client to simulate HTTP client behavior for testing."""
def init(self, base_url, should_raise=False, delay=0):
self._base_url = base_url
self.should_raise = should_raise
self.delay = delay
self.executed_requests = []

def get_base_url(self):
    return self._base_url

async def execute(self, request):
    # Simulate delay if specified
    if self.delay > 0:
        await asyncio.sleep(self.delay)
    self.executed_requests.append(request)
    if self.should_raise:
        raise RuntimeError("Simulated client error")
    # Return a dummy response with the request body for validation
    return HTTPResponse({
        "method": request.method,
        "url": request.url,
        "headers": request.headers,
        "body": request.body,
        "path_params": request.path_params,
        "query_params": request.query_params,
    })

class DummyJiraClient:
"""A dummy wrapper matching JiraClient interface."""
def init(self, dummy_client):
self.client = dummy_client

def get_client(self):
    return self.client

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

--- Test suite for JiraDataSource.assign_issue_type_screen_scheme_to_project ---

1. BASIC TEST CASES

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_basic_success():
"""Test basic successful call with both parameters."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)
resp = await datasource.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId="scheme123",
projectId="proj456"
)
data = resp.json()

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_only_scheme_id():
"""Test with only issueTypeScreenSchemeId provided."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)
resp = await datasource.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId="schemeXYZ"
)
data = resp.json()

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_only_project_id():
"""Test with only projectId provided."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)
resp = await datasource.assign_issue_type_screen_scheme_to_project(
projectId="projABC"
)
data = resp.json()

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_default_headers():
"""Test that Content-Type header is set by default."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)
resp = await datasource.assign_issue_type_screen_scheme_to_project()
data = resp.json()

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_custom_headers():
"""Test that custom headers are merged and override defaults."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)
custom_headers = {"Content-Type": "custom/type", "X-Test": "1"}
resp = await datasource.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId="foo",
projectId="bar",
headers=custom_headers
)
data = resp.json()

2. EDGE TEST CASES

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

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_client_no_base_url():
"""Test that ValueError is raised if client lacks get_base_url()."""
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_assign_issue_type_screen_scheme_to_project_client_execute_raises():
"""Test that exceptions from client.execute are propagated."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net", should_raise=True)
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)
with pytest.raises(RuntimeError, match="Simulated client error"):
await datasource.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId="fail",
projectId="fail"
)

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_none_parameters():
"""Test that passing None for all parameters works (empty body)."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)
resp = await datasource.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId=None,
projectId=None
)
data = resp.json()

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_empty_strings():
"""Test that empty strings are included in the body."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)
resp = await datasource.assign_issue_type_screen_scheme_to_project(
issueTypeScreenSchemeId="",
projectId=""
)
data = resp.json()

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_concurrent_execution():
"""Test concurrent execution of the async method."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)

async def call(i):
    return await datasource.assign_issue_type_screen_scheme_to_project(
        issueTypeScreenSchemeId=f"scheme{i}",
        projectId=f"proj{i}"
    )

results = await asyncio.gather(*(call(i) for i in range(5)))
for i, resp in enumerate(results):
    data = resp.json()

3. LARGE SCALE TEST CASES

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_large_scale():
"""Test with a large number of concurrent requests (but <100)."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)

async def call(i):
    return await datasource.assign_issue_type_screen_scheme_to_project(
        issueTypeScreenSchemeId=f"scheme{i}",
        projectId=f"proj{i}"
    )

N = 50  # keep under 100 for fast test
results = await asyncio.gather(*(call(i) for i in range(N)))
for i, resp in enumerate(results):
    data = resp.json()

4. THROUGHPUT TEST CASES

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_throughput_small_load():
"""Throughput test: small load, 5 concurrent calls."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)

async def call(i):
    return await datasource.assign_issue_type_screen_scheme_to_project(
        issueTypeScreenSchemeId=f"scheme{i}",
        projectId=f"proj{i}"
    )

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

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_throughput_medium_load():
"""Throughput test: medium load, 20 concurrent calls."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)

async def call(i):
    return await datasource.assign_issue_type_screen_scheme_to_project(
        issueTypeScreenSchemeId=f"scheme{i}",
        projectId=f"proj{i}"
    )

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

@pytest.mark.asyncio
async def test_assign_issue_type_screen_scheme_to_project_throughput_high_volume():
"""Throughput test: high volume, 100 concurrent calls."""
dummy_client = DummyAsyncClient(base_url="https://example.atlassian.net")
jira_client = DummyJiraClient(dummy_client)
datasource = JiraDataSource(jira_client)

async def call(i):
    return await datasource.assign_issue_type_screen_scheme_to_project(
        issueTypeScreenSchemeId=f"scheme{i}",
        projectId=f"proj{i}"
    )

N = 100  # upper bound for fast, deterministic test
results = await asyncio.gather(*(call(i) for i in range(N)))
for resp in 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.assign_issue_type_screen_scheme_to_project-mhsbp24m and push.