⚡️ Speed up method ConfluenceDataSource.get_custom_content_content_properties_by_id by 19%

[codeflash-ai]

📄 19% (0.19x) speedup for ConfluenceDataSource.get_custom_content_content_properties_by_id in backend/python/app/sources/external/confluence/confluence.py

⏱️ Runtime : 4.07 milliseconds → 3.42 milliseconds (best of 250 runs)

📝 Explanation and details

The optimization achieves a 19% runtime speedup and 13.1% throughput improvement through two key micro-optimizations that reduce unnecessary object allocations and conditional overhead:

Key Optimizations

1. Conditional Headers Processing

• Original: dict(headers or {}) - Always creates a new dict, even when headers is None
• Optimized: dict(headers) if headers else {} - Only creates dict when headers exists, uses empty dict literal for None case
• Impact: Eliminates unnecessary dict construction in the common case where headers is None

2. Improved Dictionary Iteration in _as_str_dict

• Original: (d or {}).items() - Creates empty dict when d is None, then iterates
• Optimized: (d.items() if d else []) - Direct conditional returns empty list for None, avoiding dict creation
• Impact: Reduces object allocation overhead when processing empty dictionaries

Performance Analysis

The line profiler shows significant improvements:

• _headers assignment time reduced from 501.2ns to 262ns per hit (48% faster)
• _as_str_dict function time reduced from 2284.4ns to 1794.4ns per hit (21% faster)
• Overall function execution improved from 13.77ms to 11.20ms

Test Case Performance

The optimizations are particularly effective for:

• Basic usage patterns where headers are frequently None or empty
• High-throughput scenarios with many concurrent requests (50-100 calls)
• Sustained execution patterns with repeated API calls

These micro-optimizations compound when the function is called frequently in API request workflows, making them valuable for high-frequency Confluence API operations where even small per-call improvements translate to meaningful throughput gains.

✅ Correctness verification report:

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

🌀 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.confluence.confluence import ConfluenceDataSource

---- 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, response_content):
self.content = response_content

---- Minimal stub/mock for HTTP client ----

class DummyAsyncHTTPClient:
"""A dummy async HTTP client to simulate execute() and get_base_url()."""
def init(self, base_url="https://dummy.atlassian.net"):
self._base_url = base_url
self.last_request = None
self.execute_call_count = 0

def get_base_url(self):
    return self._base_url

async def execute(self, req):
    self.last_request = req
    self.execute_call_count += 1
    # Simulate a response with the request info for validation
    return HTTPResponse({
        "method": req.method,
        "url": req.url,
        "headers": req.headers,
        "path_params": req.path_params,
        "query_params": req.query_params,
        "body": req.body,
    })

---- Minimal stub for ConfluenceClient ----

class DummyConfluenceClient:
def init(self, dummy_http_client=None):
self._dummy_http_client = dummy_http_client or DummyAsyncHTTPClient()
def get_client(self):
return self._dummy_http_client
from app.sources.external.confluence.confluence import ConfluenceDataSource

---- TESTS ----

--- 1. Basic Test Cases ---

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_basic():
"""Test that the function returns expected HTTPResponse content for normal inputs."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
custom_content_id = 123
property_id = 456
headers = {"X-Test": "abc"}
resp = await ds.get_custom_content_content_properties_by_id(custom_content_id, property_id, headers)

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_no_headers():
"""Test that the function works when headers is None."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_content_properties_by_id(1, 2)

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_empty_headers():
"""Test that the function works when headers is an empty dict."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_content_properties_by_id(3, 4, {})

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_string_headers():
"""Test that the function serializes header values to strings."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_content_properties_by_id(10, 20, {"X-Num": 123, "X-Bool": True})

--- 2. Edge Test Cases ---

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_negative_ids():
"""Test with negative custom_content_id and property_id."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_content_properties_by_id(-1, -999)

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_zero_ids():
"""Test with 0 as custom_content_id and property_id."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_content_properties_by_id(0, 0)

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_large_ids():
"""Test with very large integer IDs."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
large_id = 2**62
resp = await ds.get_custom_content_content_properties_by_id(large_id, large_id)

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

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_client_missing_get_base_url():
"""Test that ValueError is raised if client lacks 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"):
ConfluenceDataSource(NoBaseUrlClient())

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_concurrent_execution():
"""Test concurrent execution of the async function with different parameters."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
# Launch several concurrent calls with different IDs
ids = [(1, 10), (2, 20), (3, 30), (4, 40)]
results = await asyncio.gather(
*(ds.get_custom_content_content_properties_by_id(cid, pid, {"X": idx}) for idx, (cid, pid) in enumerate(ids))
)
for idx, (cid, pid) in enumerate(ids):
resp = results[idx]

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_headers_various_types():
"""Test that headers with various value types are stringified."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
headers = {
"X-Int": 1,
"X-Float": 3.14,
"X-Bool": False,
"X-None": None,
"X-List": [1, 2, 3]
}
resp = await ds.get_custom_content_content_properties_by_id(1, 2, headers)

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

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_many_concurrent():
"""Test the function under a moderate concurrent load (50 concurrent calls)."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
num_calls = 50
tasks = [
ds.get_custom_content_content_properties_by_id(i, i+100, {"X-Index": i})
for i in range(num_calls)
]
results = await asyncio.gather(*tasks)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_url_formatting_edge():
"""Test with IDs that could cause URL formatting issues."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
# IDs with special characters when cast to string (simulate edge, even though IDs are int)
resp = await ds.get_custom_content_content_properties_by_id(0, -1)

--- 4. Throughput Test Cases ---

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_throughput_small_load():
"""Throughput: Test function performance with a small burst of calls."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
num_calls = 10
tasks = [
ds.get_custom_content_content_properties_by_id(i, i+1)
for i in range(num_calls)
]
results = await asyncio.gather(*tasks)
# Ensure all responses are correct and unique
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_throughput_medium_load():
"""Throughput: Test function with a medium volume of requests."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
num_calls = 100
tasks = [
ds.get_custom_content_content_properties_by_id(i, 1000-i)
for i in range(num_calls)
]
results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_throughput_headers_variation():
"""Throughput: Test with many concurrent calls, each with different headers."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
num_calls = 20
tasks = [
ds.get_custom_content_content_properties_by_id(i, i+2, {"X-Seq": f"seq-{i}"})
for i in range(num_calls)
]
results = await asyncio.gather(*tasks)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_throughput_sustained_pattern():
"""Throughput: Test sustained execution pattern with repeated calls."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
# Call the function 5 times in a row for the same IDs, simulating sustained use
for i in range(5):
resp = await ds.get_custom_content_content_properties_by_id(111, 222, {"X-Iter": i})

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.confluence.confluence import ConfluenceDataSource

--- Minimal stubs and helpers for required classes ---

class DummyHTTPResponse:
"""A simple dummy HTTPResponse for testing."""
def init(self, status_code=200, data=None):
self.status_code = status_code
self.data = data or {}

def __eq__(self, other):
    if not isinstance(other, DummyHTTPResponse):
        return False
    return self.status_code == other.status_code and self.data == other.data

class DummyAsyncClient:
"""A dummy async client that mimics the required interface."""
def init(self, base_url="https://dummy.confluence.local", execute_behavior=None):
self._base_url = base_url
self.execute_behavior = execute_behavior or self.default_execute

def get_base_url(self):
    return self._base_url

async def execute(self, req):
    return await self.execute_behavior(req)

async def default_execute(self, req):
    # Return a dummy HTTPResponse with the request URL for verification
    return DummyHTTPResponse(status_code=200, data={"url": req.url, "method": req.method, "headers": req.headers, "path_params": req.path_params, "query_params": req.query_params, "body": req.body})

class DummyConfluenceClient:
"""A dummy ConfluenceClient that returns a dummy async client."""
def init(self, client=None):
self.client = client or DummyAsyncClient()

def get_client(self):
    return self.client

--- Required request/response classes from the code under test ---

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(DummyHTTPResponse):
pass
from app.sources.external.confluence.confluence import ConfluenceDataSource

--- TESTS ---

1. Basic Test Cases

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_basic():
"""Test that the function returns a DummyHTTPResponse with correct URL and method."""
dummy_client = DummyConfluenceClient()
ds = ConfluenceDataSource(dummy_client)
custom_content_id = 123
property_id = 456
resp = await ds.get_custom_content_content_properties_by_id(custom_content_id, property_id)
# The URL should be correctly formatted
expected_url = "https://dummy.confluence.local/custom-content/123/properties/456"

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_with_headers():
"""Test that custom headers are passed and stringified."""
dummy_client = DummyConfluenceClient()
ds = ConfluenceDataSource(dummy_client)
headers = {"X-Test-Header": "foo", "X-Int": 42}
resp = await ds.get_custom_content_content_properties_by_id(1, 2, headers=headers)

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_strict_types():
"""Test that integer IDs are stringified in path params."""
dummy_client = DummyConfluenceClient()
ds = ConfluenceDataSource(dummy_client)
resp = await ds.get_custom_content_content_properties_by_id(99, 88)

2. Edge Test Cases

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

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_invalid_client_missing_method():
"""Test that ValueError is raised if client does not have get_base_url."""
class NoBaseUrlClient:
def get_client(self):
class C:
pass
return C()
with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
ConfluenceDataSource(NoBaseUrlClient())

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_execute_raises():
"""Test that exceptions from execute are propagated."""
async def bad_execute(req):
raise RuntimeError("execute failed")
class FailingAsyncClient(DummyAsyncClient):
async def execute(self, req):
return await bad_execute(req)
dummy_client = DummyConfluenceClient(client=FailingAsyncClient())
ds = ConfluenceDataSource(dummy_client)
with pytest.raises(RuntimeError, match="execute failed"):
await ds.get_custom_content_content_properties_by_id(1, 2)

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_concurrent():
"""Test concurrent execution of the async function."""
dummy_client = DummyConfluenceClient()
ds = ConfluenceDataSource(dummy_client)
# Run 10 concurrent requests with different IDs
results = await asyncio.gather(
*[ds.get_custom_content_content_properties_by_id(i, i+100) for i in range(10)]
)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_empty_headers():
"""Test that passing headers=None or headers={} works identically."""
dummy_client = DummyConfluenceClient()
ds = ConfluenceDataSource(dummy_client)
resp1 = await ds.get_custom_content_content_properties_by_id(1, 2, headers=None)
resp2 = await ds.get_custom_content_content_properties_by_id(1, 2, headers={})

3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_large_scale_concurrent():
"""Test concurrent execution under moderate load (50 requests)."""
dummy_client = DummyConfluenceClient()
ds = ConfluenceDataSource(dummy_client)
N = 50
ids = list(range(1000, 1000 + N))
results = await asyncio.gather(
*[ds.get_custom_content_content_properties_by_id(cid, cid+1) for cid in ids]
)
for i, resp in enumerate(results):
expected_url = f"https://dummy.confluence.local/custom-content/{ids[i]}/properties/{ids[i]+1}"

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_throughput_small_load():
"""Throughput: Test the function under small load (5 concurrent calls)."""
dummy_client = DummyConfluenceClient()
ds = ConfluenceDataSource(dummy_client)
ids = list(range(5))
results = await asyncio.gather(
*[ds.get_custom_content_content_properties_by_id(i, i+10) for i in ids]
)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_throughput_medium_load():
"""Throughput: Test the function under medium load (25 concurrent calls)."""
dummy_client = DummyConfluenceClient()
ds = ConfluenceDataSource(dummy_client)
ids = list(range(25))
results = await asyncio.gather(
*[ds.get_custom_content_content_properties_by_id(i, i+20) for i in ids]
)
for i, resp in enumerate(results):
pass

@pytest.mark.asyncio
async def test_get_custom_content_content_properties_by_id_throughput_large_load():
"""Throughput: Test the function under large but reasonable load (100 concurrent calls)."""
dummy_client = DummyConfluenceClient()
ds = ConfluenceDataSource(dummy_client)
ids = list(range(100))
results = await asyncio.gather(
*[ds.get_custom_content_content_properties_by_id(i, i+1000) for i in ids]
)
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.

To edit these changes git checkout codeflash/optimize-ConfluenceDataSource.get_custom_content_content_properties_by_id-mhvfdmhx and push.