⚡️ Speed up method JiraDataSource.migration_resource_workflow_rule_search_post by 10%

[codeflash-ai]

📄 10% (0.10x) speedup for JiraDataSource.migration_resource_workflow_rule_search_post in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 2.16 milliseconds → 1.96 milliseconds (best of 51 runs)

📝 Explanation and details

The optimization achieves a 10% runtime improvement by implementing a simple but effective early-return check in the _safe_format_url function.

Key optimization applied:

The optimized version adds an early check if not params: at the beginning of _safe_format_url. When the params dictionary is empty (which is the case in this workload where _path is always an empty dict), the function immediately returns the template string without executing the expensive template.format_map(_SafeDict(params)) operation.

Why this delivers speedup:

From the line profiler data, _safe_format_url time dropped from 518,729 ns to 149,302 ns (71% reduction). The original version always created a _SafeDict object and called format_map(), then caught the exception and returned the template anyway. The optimized version skips this entirely when no formatting is needed.

This optimization is particularly effective because:

• The function is called 317 times in the test run
• The _path parameter is always empty in this HTTP POST endpoint
• Exception handling (try/except) adds overhead even when no actual formatting occurs
• Object creation (_SafeDict(params)) is avoided entirely

Impact on workloads:

The 10% runtime improvement will benefit any application making frequent Jira API calls through this data source, especially in batch operations or high-throughput scenarios. The optimization is most effective when path parameters are not used (common for POST endpoints), making it well-suited for workflow management and bulk operations against Jira's REST API.

Note that while runtime improved by 10%, throughput remained constant at 16,167 ops/sec, indicating the bottleneck likely shifts to other parts of the async execution pipeline after this optimization.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio

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

# --- Minimal stubs for HTTPRequest and HTTPResponse to allow 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


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


# --- Minimal stubs for JiraClient and HTTPClient ---


class DummyHTTPClient:
    """A dummy async HTTP client for testing."""

    def __init__(self, base_url):
        self._base_url = base_url
        self.last_request = None
        self.should_raise = None
        self.response_data = None
        self.delay = 0  # For simulating async

    def get_base_url(self):
        return self._base_url

    async def execute(self, req):
        self.last_request = req
        if self.should_raise:
            raise self.should_raise
        if self.delay:
            await asyncio.sleep(self.delay)
        # Simulate a response echoing the request
        return HTTPResponse(
            {
                "method": req.method,
                "url": req.url,
                "headers": req.headers,
                "body": req.body,
            }
        )


class DummyJiraClient:
    def __init__(self, base_url):
        self._client = DummyHTTPClient(base_url)

    def get_client(self):
        return self._client


# --- TESTS ---


@pytest.mark.asyncio
async def test_basic_post_and_response_fields():
    """Basic: Test normal async call returns expected HTTPResponse data."""
    client = DummyJiraClient("https://example.com")
    ds = JiraDataSource(client)
    resp = await ds.migration_resource_workflow_rule_search_post(
        Atlassian_Transfer_Id="abc123", ruleIds=["r1", "r2"], workflowEntityId="wf_42"
    )


@pytest.mark.asyncio
async def test_basic_with_expand_and_headers():
    """Basic: Test expand and custom headers are handled and merged."""
    client = DummyJiraClient("https://jira.local")
    ds = JiraDataSource(client)
    resp = await ds.migration_resource_workflow_rule_search_post(
        Atlassian_Transfer_Id="tid",
        ruleIds=["x"],
        workflowEntityId="wid",
        expand="details",
        headers={"X-Test": "yes"},
    )


@pytest.mark.asyncio
async def test_basic_async_behavior():
    """Basic: Awaiting the coroutine yields the result."""
    client = DummyJiraClient("https://jira.async")
    ds = JiraDataSource(client)
    codeflash_output = ds.migration_resource_workflow_rule_search_post("T", ["A"], "W")
    coro = codeflash_output
    result = await coro


@pytest.mark.asyncio
async def test_edge_empty_ruleids_and_expand_none():
    """Edge: ruleIds empty, expand is None."""
    client = DummyJiraClient("https://jira.edge")
    ds = JiraDataSource(client)
    resp = await ds.migration_resource_workflow_rule_search_post(
        Atlassian_Transfer_Id="edge", ruleIds=[], workflowEntityId="wf", expand=None
    )


@pytest.mark.asyncio
async def test_edge_headers_override_content_type():
    """Edge: Custom Content-Type header is not overridden if provided."""
    client = DummyJiraClient("https://jira.headers")
    ds = JiraDataSource(client)
    resp = await ds.migration_resource_workflow_rule_search_post(
        Atlassian_Transfer_Id="h1",
        ruleIds=["r"],
        workflowEntityId="w",
        headers={"Content-Type": "application/custom+json"},
    )


@pytest.mark.asyncio
async def test_edge_concurrent_execution():
    """Edge: Multiple concurrent async calls do not interfere."""
    client = DummyJiraClient("https://jira.concurrent")
    ds = JiraDataSource(client)
    tasks = [
        ds.migration_resource_workflow_rule_search_post(f"id{i}", [f"r{i}"], f"wf{i}")
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_edge_http_client_not_initialized():
    """Edge: Raises ValueError if HTTP 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_edge_http_client_missing_get_base_url():
    """Edge: Raises ValueError if get_base_url is missing."""

    class BadHTTPClient:
        pass

    class BadClient:
        def get_client(self):
            return BadHTTPClient()

    with pytest.raises(
        ValueError, match="HTTP client does not have get_base_url method"
    ):
        JiraDataSource(BadClient())


@pytest.mark.asyncio
async def test_edge_http_client_execute_raises():
    """Edge: If execute raises, the exception propagates."""

    class FailingHTTPClient(DummyHTTPClient):
        async def execute(self, req):
            raise RuntimeError("fail!")

    class FailingClient:
        def get_client(self):
            return FailingHTTPClient("https://fail")

    ds = JiraDataSource(FailingClient())
    with pytest.raises(RuntimeError, match="fail!"):
        await ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id="fail", ruleIds=["x"], workflowEntityId="w"
        )


@pytest.mark.asyncio
async def test_large_scale_many_concurrent_requests():
    """Large Scale: 50 concurrent requests."""
    client = DummyJiraClient("https://jira.large")
    ds = JiraDataSource(client)
    N = 50
    tasks = [
        ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"tid{i}", ruleIds=[str(i)], workflowEntityId=f"wf{i}"
        )
        for i in range(N)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_large_scale_varied_inputs():
    """Large Scale: Varied ruleIds and workflowEntityId types."""
    client = DummyJiraClient("https://jira.varied")
    ds = JiraDataSource(client)
    resp = await ds.migration_resource_workflow_rule_search_post(
        Atlassian_Transfer_Id="varied",
        ruleIds=["r1", "r2", "r3"],
        workflowEntityId="wf_varied",
        expand="all",
    )


@pytest.mark.asyncio
async def test_JiraDataSource_migration_resource_workflow_rule_search_post_throughput_small_load():
    """Throughput: Small load, 5 sequential calls."""
    client = DummyJiraClient("https://jira.throughput")
    ds = JiraDataSource(client)
    for i in range(5):
        resp = await ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"t{i}", ruleIds=[f"r{i}"], workflowEntityId=f"wf{i}"
        )


@pytest.mark.asyncio
async def test_JiraDataSource_migration_resource_workflow_rule_search_post_throughput_medium_load():
    """Throughput: Medium load, 20 concurrent calls."""
    client = DummyJiraClient("https://jira.throughput.medium")
    ds = JiraDataSource(client)
    N = 20
    tasks = [
        ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"tid{i}",
            ruleIds=[f"r{i}"],
            workflowEntityId=f"wf{i}",
        )
        for i in range(N)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_JiraDataSource_migration_resource_workflow_rule_search_post_throughput_large_load():
    """Throughput: Large load, 100 concurrent calls."""
    client = DummyJiraClient("https://jira.throughput.large")
    ds = JiraDataSource(client)
    N = 100
    tasks = [
        ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"tid{i}",
            ruleIds=[f"r{i}"],
            workflowEntityId=f"wf{i}",
        )
        for i in range(N)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_JiraDataSource_migration_resource_workflow_rule_search_post_throughput_sustained_pattern():
    """Throughput: Sustained pattern, alternating concurrent and sequential."""
    client = DummyJiraClient("https://jira.throughput.sustained")
    ds = JiraDataSource(client)
    # 10 sequential
    for i in range(10):
        resp = await ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"seq{i}",
            ruleIds=[f"r{i}"],
            workflowEntityId=f"wf{i}",
        )
    # 10 concurrent
    tasks = [
        ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"con{i}",
            ruleIds=[f"r{i}"],
            workflowEntityId=f"wf{i}",
        )
        for i in range(10)
    ]
    results = await asyncio.gather(*tasks)
    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

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

# --- Minimal stubs for required classes (to avoid external dependencies) ---


class HTTPRequest:
    def __init__(
        self,
        method: str,
        url: str,
        headers: Dict[str, str],
        path_params: Dict[str, str],
        query_params: Dict[str, str],
        body: Any,
    ):
        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,
        content: Any,
        status_code: int = 200,
        headers: Optional[Dict[str, str]] = None,
    ):
        self.content = content
        self.status_code = status_code
        self.headers = headers or {}

    def json(self):
        return self.content


class DummyAsyncClient:
    """Dummy async client to simulate HTTPClient behavior."""

    def __init__(self, base_url: str):
        self._base_url = base_url

    def get_base_url(self):
        return self._base_url

    async def execute(self, request: HTTPRequest):
        # Simulate a response based on input
        # For testing, echo back the request body and headers
        return HTTPResponse(
            content={
                "body": request.body,
                "headers": request.headers,
                "url": request.url,
                "method": request.method,
            },
            status_code=200,
            headers=request.headers,
        )


class JiraClient:
    def __init__(self, client: DummyAsyncClient):
        self.client = client

    def get_client(self):
        return self.client


def codeflash_capture(*args, **kwargs):
    # Dummy decorator for compatibility
    def decorator(fn):
        return fn

    return decorator


# --- Unit Tests ---


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_basic():
    """Basic test: normal input, all required fields."""
    client = JiraClient(DummyAsyncClient(base_url="https://jira.example.com"))
    ds = JiraDataSource(client)
    response = await ds.migration_resource_workflow_rule_search_post(
        Atlassian_Transfer_Id="transfer-123",
        ruleIds=["rule1", "rule2"],
        workflowEntityId="wf-entity-42",
    )
    content = response.json()


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_with_expand_and_headers():
    """Test with optional expand and extra headers."""
    client = JiraClient(DummyAsyncClient(base_url="https://jira.example.com"))
    ds = JiraDataSource(client)
    response = await ds.migration_resource_workflow_rule_search_post(
        Atlassian_Transfer_Id="transfer-456",
        ruleIds=["ruleA"],
        workflowEntityId="wf-entity-99",
        expand="details",
        headers={"X-Custom": "value"},
    )
    content = response.json()


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_empty_ruleIds():
    """Edge case: empty ruleIds list."""
    client = JiraClient(DummyAsyncClient(base_url="https://jira.example.com"))
    ds = JiraDataSource(client)
    response = await ds.migration_resource_workflow_rule_search_post(
        Atlassian_Transfer_Id="transfer-empty", ruleIds=[], workflowEntityId="wf-empty"
    )
    content = response.json()


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_none_expand_and_headers():
    """Edge case: expand and headers are None."""
    client = JiraClient(DummyAsyncClient(base_url="https://jira.example.com"))
    ds = JiraDataSource(client)
    response = await ds.migration_resource_workflow_rule_search_post(
        Atlassian_Transfer_Id="transfer-none",
        ruleIds=["ruleX"],
        workflowEntityId="wf-none",
        expand=None,
        headers=None,
    )
    content = response.json()


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_concurrent_execution():
    """Edge case: concurrent execution of multiple requests."""
    client = JiraClient(DummyAsyncClient(base_url="https://jira.example.com"))
    ds = JiraDataSource(client)
    coros = [
        ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"transfer-{i}",
            ruleIds=[f"rule{i}"],
            workflowEntityId=f"wf-{i}",
        )
        for i in range(5)
    ]
    responses = await asyncio.gather(*coros)
    for i, response in enumerate(responses):
        content = response.json()


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_invalid_client():
    """Edge case: HTTP client is None, should raise ValueError."""

    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_migration_resource_workflow_rule_search_post_invalid_base_url():
    """Edge case: client missing get_base_url, should raise ValueError."""

    class BadClient:
        def get_client(self):
            return object()

    with pytest.raises(
        ValueError, match="HTTP client does not have get_base_url method"
    ):
        JiraDataSource(BadClient())


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_large_ruleIds():
    """Large scale: many ruleIds in the request."""
    client = JiraClient(DummyAsyncClient(base_url="https://jira.example.com"))
    ds = JiraDataSource(client)
    ruleIds = [f"rule{i}" for i in range(100)]
    response = await ds.migration_resource_workflow_rule_search_post(
        Atlassian_Transfer_Id="transfer-large",
        ruleIds=ruleIds,
        workflowEntityId="wf-large",
    )
    content = response.json()


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_concurrent_large_scale():
    """Large scale: concurrent execution with large ruleIds lists."""
    client = JiraClient(DummyAsyncClient(base_url="https://jira.example.com"))
    ds = JiraDataSource(client)
    coros = [
        ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"transfer-{i}",
            ruleIds=[f"rule{i}-{j}" for j in range(50)],
            workflowEntityId=f"wf-{i}",
        )
        for i in range(10)
    ]
    responses = await asyncio.gather(*coros)
    for i, response in enumerate(responses):
        content = response.json()


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_throughput_small_load():
    """Throughput test: small load, multiple sequential requests."""
    client = JiraClient(DummyAsyncClient(base_url="https://jira.example.com"))
    ds = JiraDataSource(client)
    for i in range(10):
        response = await ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"transfer-{i}",
            ruleIds=[f"rule{i}"],
            workflowEntityId=f"wf-{i}",
        )
        content = response.json()


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_throughput_medium_load():
    """Throughput test: medium load, concurrent requests."""
    client = JiraClient(DummyAsyncClient(base_url="https://jira.example.com"))
    ds = JiraDataSource(client)
    coros = [
        ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"transfer-{i}",
            ruleIds=[f"rule{i}"],
            workflowEntityId=f"wf-{i}",
        )
        for i in range(30)
    ]
    responses = await asyncio.gather(*coros)
    for i, response in enumerate(responses):
        content = response.json()


@pytest.mark.asyncio
async def test_migration_resource_workflow_rule_search_post_throughput_high_volume():
    """Throughput test: high volume, concurrent requests, moderate data size."""
    client = JiraClient(DummyAsyncClient(base_url="https://jira.example.com"))
    ds = JiraDataSource(client)
    coros = [
        ds.migration_resource_workflow_rule_search_post(
            Atlassian_Transfer_Id=f"transfer-{i}",
            ruleIds=[f"rule{i}-{j}" for j in range(10)],
            workflowEntityId=f"wf-{i}",
        )
        for i in range(50)
    ]
    responses = await asyncio.gather(*coros)
    for i, response in enumerate(responses):
        content = response.json()


# 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.migration_resource_workflow_rule_search_post-miqso24x and push.