⚡️ Speed up method JiraDataSource.create_dashboard by 22%

[codeflash-ai]

📄 22% (0.22x) speedup for JiraDataSource.create_dashboard in backend/python/app/sources/external/jira/jira.py

⏱️ Runtime : 2.10 milliseconds → 1.73 milliseconds (best of 245 runs)

📝 Explanation and details

The optimization achieves a 21% runtime improvement by targeting the most expensive helper functions called during HTTP request preparation:

Key Optimizations:

1. _as_str_dict Early Returns: Added two fast-path checks that avoid expensive operations:

   • Empty dict check (if not d: return {}) - saves 15.1% of function time
   • All-strings check (if all(isinstance(k, str) and isinstance(v, str)...)) - directly returns the dict when no conversion is needed, saving 54.1% of function time

2. _safe_format_url Simplification: Added if not params: return template check that completely bypasses the expensive template.format_map(_SafeDict(params)) operation when no path parameters exist (which is the common case in this API).

Performance Impact Analysis:

From the line profiler data, the original _as_str_dict spent 100% of its time (1.6ms) on dictionary comprehension for every call. The optimized version shows dramatic improvements:

• 764 calls now return early with empty dicts (15.1% of time)
• 383 calls return the original dict directly when all values are strings (7.6% of time)
• Only 2 calls actually need the expensive conversion (0.8% of time)

Similarly, _safe_format_url went from 566μs total time with expensive format operations to just 173μs by early-returning the template string.

Throughput Benefits: The 1.7% throughput improvement (92,303 → 93,835 ops/sec) demonstrates that these optimizations are particularly effective for high-frequency API calls where the same request patterns are repeated. The optimizations work best when:

• Path parameters are empty (very common)
• Headers/query params are already strings (typical case)
• Multiple concurrent requests use similar parameter patterns

These micro-optimizations compound significantly in HTTP client libraries where request preparation happens on every API call.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 411 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
from typing import Any, Dict, Optional

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

class DummyAsyncClient:
    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: HTTPRequest):
        # Simulate a response containing the request data for verification
        self.executed_requests.append(request)
        # For edge/error cases, simulate error if name is "error"
        if request.body.get("name") == "error":
            raise RuntimeError("Simulated error")
        return HTTPResponse({
            "method": request.method,
            "url": request.url,
            "headers": request.headers,
            "path_params": request.path_params,
            "query_params": request.query_params,
            "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

# ---- TESTS ----

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_create_dashboard_basic_success():
    """Test basic successful dashboard creation with required fields."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    edit_permissions = [{"type": "user", "user": "alice"}]
    share_permissions = [{"type": "group", "group": "devs"}]
    name = "My Dashboard"
    resp = await ds.create_dashboard(edit_permissions, name, share_permissions)

@pytest.mark.asyncio
async def test_create_dashboard_with_optional_fields():
    """Test dashboard creation including all optional fields."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com/"))
    ds = JiraDataSource(client)
    edit_permissions = [{"type": "user", "user": "bob"}]
    share_permissions = [{"type": "group", "group": "qa"}]
    name = "QA Dashboard"
    description = "Dashboard for QA team"
    headers = {"X-Custom": "value"}
    resp = await ds.create_dashboard(
        edit_permissions, name, share_permissions,
        extendAdminPermissions=True,
        description=description,
        headers=headers
    )

@pytest.mark.asyncio
async def test_create_dashboard_empty_permissions_lists():
    """Test dashboard creation with empty edit and share permissions lists."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    resp = await ds.create_dashboard([], "EmptyPerms", [])

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_create_dashboard_missing_client_raises():
    """Test that initializing JiraDataSource with None client raises ValueError."""
    class DummyNoneClient:
        def get_client(self):
            return None
    with pytest.raises(ValueError) as excinfo:
        JiraDataSource(DummyNoneClient())

@pytest.mark.asyncio
async def test_create_dashboard_client_missing_base_url_method_raises():
    """Test JiraDataSource raises if client lacks get_base_url method."""
    class DummyBadClient:
        def get_client(self):
            return object()  # No get_base_url method
    with pytest.raises(ValueError) as excinfo:
        JiraDataSource(DummyBadClient())

@pytest.mark.asyncio
async def test_create_dashboard_execute_raises_exception():
    """Test that if client.execute raises, the exception propagates."""
    class ErrorAsyncClient(DummyAsyncClient):
        async def execute(self, request):
            raise RuntimeError("Simulated error")
    client = JiraClient(ErrorAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    with pytest.raises(RuntimeError) as excinfo:
        await ds.create_dashboard([{"type": "user", "user": "alice"}], "error", [{"type": "group", "group": "devs"}])

@pytest.mark.asyncio
async def test_create_dashboard_concurrent_execution():
    """Test concurrent execution of create_dashboard to ensure async correctness."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    # Prepare several concurrent requests
    tasks = [
        ds.create_dashboard([{"type": "user", "user": f"user{i}"}], f"name{i}", [{"type": "group", "group": f"group{i}"}])
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_create_dashboard_custom_headers_override():
    """Test that custom headers override default Content-Type."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    headers = {"Content-Type": "custom/type", "X-Extra": "foo"}
    resp = await ds.create_dashboard([{"type": "user", "user": "bob"}], "CustomHeader", [{"type": "group", "group": "qa"}], headers=headers)

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_create_dashboard_many_concurrent():
    """Test large number of concurrent create_dashboard calls."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    num_tasks = 50  # Large but reasonable for unit test
    tasks = [
        ds.create_dashboard([{"type": "user", "user": f"user{i}"}], f"name{i}", [{"type": "group", "group": f"group{i}"}])
        for i in range(num_tasks)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_create_dashboard_large_permission_lists():
    """Test handling of large editPermissions and sharePermissions lists."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    edit_permissions = [{"type": "user", "user": f"user{i}"} for i in range(200)]
    share_permissions = [{"type": "group", "group": f"group{i}"} for i in range(200)]
    resp = await ds.create_dashboard(edit_permissions, "BigPerms", share_permissions)

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_create_dashboard_throughput_small_load():
    """Throughput test: small load of concurrent requests."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    tasks = [
        ds.create_dashboard([{"type": "user", "user": "small"}], "small", [{"type": "group", "group": "small"}])
        for _ in range(5)
    ]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_create_dashboard_throughput_medium_load():
    """Throughput test: medium load of concurrent requests."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    tasks = [
        ds.create_dashboard([{"type": "user", "user": f"user{i}"}], f"name{i}", [{"type": "group", "group": f"group{i}"}])
        for i in range(20)
    ]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_create_dashboard_throughput_large_load():
    """Throughput test: large load of concurrent requests."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    num_tasks = 100
    tasks = [
        ds.create_dashboard([{"type": "user", "user": f"user{i}"}], f"name{i}", [{"type": "group", "group": f"group{i}"}])
        for i in range(num_tasks)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_create_dashboard_throughput_varying_load_patterns():
    """Throughput test: varying loads and data patterns."""
    client = JiraClient(DummyAsyncClient("https://jira.example.com"))
    ds = JiraDataSource(client)
    tasks = []
    # Varying permission list lengths
    for i in range(10):
        edit_permissions = [{"type": "user", "user": f"user{i}"} for _ in range(i+1)]
        share_permissions = [{"type": "group", "group": f"group{i}"} for _ in range(10-i)]
        tasks.append(ds.create_dashboard(edit_permissions, f"name{i}", share_permissions))
    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 HTTPRequest/HTTPResponse for testing ----
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, json_data: Any, status_code: int = 200):
        self._json_data = json_data
        self.status_code = status_code

    def json(self):
        return self._json_data

# ---- Minimal JiraClient and HTTPClient for testing ----
class DummyHTTPClient:
    def __init__(self, base_url: str):
        self._base_url = base_url
        self.executed_requests = []

    def get_base_url(self):
        return self._base_url

    async def execute(self, req: HTTPRequest):
        # Record the request for inspection
        self.executed_requests.append(req)
        # Simulate a successful response containing the request body for verification
        return HTTPResponse({"received": req.body, "url": req.url, "headers": req.headers, "query": req.query_params}, status_code=201)

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

    def get_client(self):
        return self.client
from app.sources.external.jira.jira import JiraDataSource

# ---- Basic Test Cases ----

@pytest.mark.asyncio
async def test_create_dashboard_basic_success():
    """Test basic successful dashboard creation with required fields."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    edit_permissions = [{"type": "user", "value": "alice"}]
    share_permissions = [{"type": "group", "value": "devs"}]
    name = "My Dashboard"
    resp = await ds.create_dashboard(edit_permissions, name, share_permissions)
    data = resp.json()

@pytest.mark.asyncio
async def test_create_dashboard_with_optional_fields():
    """Test dashboard creation with all optional fields provided."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    edit_permissions = [{"type": "user", "value": "bob"}]
    share_permissions = [{"type": "group", "value": "qa"}]
    name = "QA Dashboard"
    description = "Dashboard for QA team"
    headers = {"X-Custom-Header": "value"}
    extend_admin = True
    resp = await ds.create_dashboard(
        edit_permissions, name, share_permissions,
        extendAdminPermissions=extend_admin,
        description=description,
        headers=headers
    )
    data = resp.json()

@pytest.mark.asyncio
async def test_create_dashboard_content_type_header_default():
    """Test that Content-Type header is set to application/json by default."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    resp = await ds.create_dashboard([{}], "Dash", [{}])
    data = resp.json()

# ---- Edge Test Cases ----

@pytest.mark.asyncio
async def test_create_dashboard_empty_permissions():
    """Test dashboard creation with empty edit and share permissions."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    resp = await ds.create_dashboard([], "EmptyPerms", [])
    data = resp.json()

@pytest.mark.asyncio
async def test_create_dashboard_empty_name():
    """Test dashboard creation with empty name string."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    resp = await ds.create_dashboard([{"type": "user", "value": "x"}], "", [{"type": "group", "value": "y"}])
    data = resp.json()

@pytest.mark.asyncio
async def test_create_dashboard_none_client_raises():
    """Test that ValueError is raised if 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_create_dashboard_client_missing_base_url_method():
    """Test that ValueError is raised if client lacks get_base_url method."""
    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_dashboard_concurrent_execution():
    """Test concurrent execution of create_dashboard."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    tasks = [
        ds.create_dashboard([{"type": "user", "value": str(i)}], f"Dash{i}", [{"type": "group", "value": "g"}])
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        data = resp.json()

@pytest.mark.asyncio
async def test_create_dashboard_exception_in_execute():
    """Test that exceptions in execute are propagated."""
    class FailingHTTPClient(DummyHTTPClient):
        async def execute(self, req):
            raise RuntimeError("fail!")
    class FailingJiraClient:
        def __init__(self):
            self.client = FailingHTTPClient("http://jira.example.com")
        def get_client(self):
            return self.client
    ds = JiraDataSource(FailingJiraClient())
    with pytest.raises(RuntimeError, match="fail!"):
        await ds.create_dashboard([{"type": "user", "value": "x"}], "Name", [{"type": "group", "value": "g"}])

# ---- Large Scale Test Cases ----

@pytest.mark.asyncio
async def test_create_dashboard_large_permissions():
    """Test dashboard creation with large numbers of permissions."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    edit_permissions = [{"type": "user", "value": f"user{i}"} for i in range(100)]
    share_permissions = [{"type": "group", "value": f"group{i}"} for i in range(100)]
    resp = await ds.create_dashboard(edit_permissions, "BigDash", share_permissions)
    data = resp.json()

@pytest.mark.asyncio
async def test_create_dashboard_large_concurrent():
    """Test many concurrent dashboard creations (up to 50)."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    tasks = [
        ds.create_dashboard([{"type": "user", "value": f"user{i}"}], f"Dash{i}", [{"type": "group", "value": f"group{i}"}])
        for i in range(50)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        data = resp.json()

# ---- Throughput Test Cases ----

@pytest.mark.asyncio
async def test_create_dashboard_throughput_small_load():
    """Test throughput with a small load of 5 dashboard creations."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    tasks = [
        ds.create_dashboard([{"type": "user", "value": f"user{i}"}], f"Dash{i}", [{"type": "group", "value": f"group{i}"}])
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)
    for resp in results:
        pass

@pytest.mark.asyncio
async def test_create_dashboard_throughput_medium_load():
    """Test throughput with a medium load of 20 dashboard creations."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    tasks = [
        ds.create_dashboard([{"type": "user", "value": f"user{i}"}], f"Dash{i}", [{"type": "group", "value": f"group{i}"}])
        for i in range(20)
    ]
    results = await asyncio.gather(*tasks)
    for resp in results:
        pass

@pytest.mark.asyncio
async def test_create_dashboard_throughput_high_load():
    """Test throughput with a high load of 100 dashboard creations."""
    client = DummyJiraClient("http://jira.example.com")
    ds = JiraDataSource(client)
    tasks = [
        ds.create_dashboard([{"type": "user", "value": f"user{i}"}], f"Dash{i}", [{"type": "group", "value": f"group{i}"}])
        for i in range(100)
    ]
    results = await asyncio.gather(*tasks)
    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.create_dashboard-mhp7wp2l and push.