⚡️ Speed up method ConfluenceDataSource.get_custom_content_attachments by 7%

[codeflash-ai]

📄 7% (0.07x) speedup for ConfluenceDataSource.get_custom_content_attachments in backend/python/app/sources/external/confluence/confluence.py

⏱️ Runtime : 2.71 milliseconds → 2.54 milliseconds (best of 173 runs)

📝 Explanation and details

The optimization achieves a 6% runtime improvement and 3.6% throughput increase through two key changes in the HTTP client's request handling:

Key Optimizations:

1. More efficient header merging in HTTPClient.execute(): Changed from dictionary unpacking {**self.headers, **request.headers} to using .copy() and .update() methods. This avoids creating intermediate dictionaries and reduces memory allocation overhead when merging headers.

2. Streamlined request kwargs construction: Replaced the original pattern of creating request_kwargs with unpacking (**kwargs) to building the dictionary incrementally with dict() and .update(). This reduces the number of dictionary operations and memory allocations.

Performance Impact:
The line profiler shows the optimization primarily benefits the _as_str_dict function calls (used for serializing headers, path params, and query params), with total time decreasing from 2.44ms to 2.27ms across all calls. While individual function improvements appear modest, the cumulative effect across multiple dictionary operations per request adds up.

Test Case Benefits:
The optimization is most effective for test cases that:

• Make concurrent requests (like test_get_custom_content_attachments_throughput_varied_parameters with 20 concurrent calls)
• Include multiple optional parameters requiring serialization
• Have custom headers that need merging

Since this is an HTTP client function likely called frequently in API-heavy workloads, even small per-request optimizations compound significantly under load, making the 6% improvement meaningful for throughput-sensitive applications.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio # used to run async functions

---- Function under test (copied exactly as provided) ----

from typing import Any, Dict, Optional, Union

import pytest # used for our unit tests
from app.sources.external.confluence.confluence import ConfluenceDataSource

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

class DummyHTTPResponse:
"""A minimal stub for HTTPResponse to simulate a real HTTP response."""
def init(self, data):
self.data = data

def __eq__(self, other):
    # For easy comparison in asserts
    if not isinstance(other, DummyHTTPResponse):
        return False
    return self.data == other.data

def __repr__(self):
    return f"DummyHTTPResponse({self.data!r})"

class DummyHTTPClient:
"""A dummy HTTP client that simulates async HTTP execution."""
def init(self, base_url="http://dummy"):
self._base_url = base_url
self.last_request = None
self.should_raise = False
self.response_data = None

def get_base_url(self):
    return self._base_url

async def execute(self, request):
    """Simulate async HTTP execution."""
    self.last_request = request
    if self.should_raise:
        raise RuntimeError("Simulated HTTP error")
    # Return a dummy response, echoing request for testability
    return DummyHTTPResponse({
        "method": request.method,
        "url": request.url,
        "headers": request.headers,
        "path_params": request.path_params,
        "query_params": request.query_params,
        "body": request.body,
        "custom": self.response_data,
    })

class DummyConfluenceClient:
"""A dummy ConfluenceClient that returns a DummyHTTPClient."""
def init(self, http_client=None):
self._http_client = http_client or DummyHTTPClient()

def get_client(self):
    return self._http_client

---- Minimal HTTPRequest stub ----

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
from app.sources.external.confluence.confluence import ConfluenceDataSource

---- UNIT TESTS ----

1. Basic Test Cases

@pytest.mark.asyncio
async def test_get_custom_content_attachments_basic_minimal():
"""Test basic usage with only required id argument."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_attachments(123)

@pytest.mark.asyncio
async def test_get_custom_content_attachments_basic_all_args():
"""Test passing all optional arguments to function."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_attachments(
id=999,
sort={"field": "name"},
cursor="abc123",
status=["current", "archived"],
mediaType="image/png",
filename="pic.png",
limit=10,
headers={"X-Test": "yes"}
)

@pytest.mark.asyncio
async def test_get_custom_content_attachments_basic_async_behavior():
"""Test that the async function can be awaited and returns immediately."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
# Await the coroutine and check for correct type
result = await ds.get_custom_content_attachments(1)

2. Edge Test Cases

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

@pytest.mark.asyncio
async def test_get_custom_content_attachments_edge_missing_get_base_url():
"""Test that ValueError is raised if HTTP client lacks get_base_url."""
class BadHTTPClient:
pass
class BadConfluenceClient:
def get_client(self):
return BadHTTPClient()
with pytest.raises(ValueError, match="get_base_url"):
ConfluenceDataSource(BadConfluenceClient())

@pytest.mark.asyncio
async def test_get_custom_content_attachments_edge_concurrent_execution():
"""Test concurrent execution of multiple async calls."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
# Run 5 concurrent calls with different ids
results = await asyncio.gather(
ds.get_custom_content_attachments(1),
ds.get_custom_content_attachments(2),
ds.get_custom_content_attachments(3),
ds.get_custom_content_attachments(4),
ds.get_custom_content_attachments(5),
)

@pytest.mark.asyncio
async def test_get_custom_content_attachments_edge_http_execute_raises():
"""Test that exceptions in the underlying HTTP client are propagated."""
client = DummyHTTPClient()
client.should_raise = True
confluence_client = DummyConfluenceClient(client)
ds = ConfluenceDataSource(confluence_client)
with pytest.raises(RuntimeError, match="Simulated HTTP error"):
await ds.get_custom_content_attachments(42)

@pytest.mark.asyncio
async def test_get_custom_content_attachments_edge_unusual_types():
"""Test edge cases with unusual types for optional arguments."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
# status as empty list, sort as empty dict, filename as empty string, limit as 0
resp = await ds.get_custom_content_attachments(
id=7,
sort={},
cursor=None,
status=[],
mediaType=None,
filename="",
limit=0,
headers=None
)

3. Large Scale Test Cases

@pytest.mark.asyncio

async def test_get_custom_content_attachments_large_scale_varied_args():
"""Test concurrent calls with varied arguments."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
coros = [
ds.get_custom_content_attachments(
id=i,
sort={"s": i},
cursor=str(i) if i % 2 == 0 else None,
status=["current"] if i % 3 == 0 else None,
mediaType="type" if i % 5 == 0 else None,
filename=f"file{i}.txt" if i % 7 == 0 else None,
limit=i if i % 11 == 0 else None,
headers={"X-Req": str(i)} if i % 13 == 0 else None
)
for i in range(20, 40)
]
results = await asyncio.gather(*coros)
# Spot check a few
for i, resp in zip(range(20, 40), results):
pass

4. Throughput Test Cases

@pytest.mark.asyncio
async def test_get_custom_content_attachments_throughput_small_load():
"""Throughput test: 10 concurrent requests (small load)."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
coros = [ds.get_custom_content_attachments(i) for i in range(10)]
results = await asyncio.gather(*coros)

@pytest.mark.asyncio

async def test_get_custom_content_attachments_throughput_varied_load():
"""Throughput test: 20 requests with varied optional arguments."""
client = DummyConfluenceClient()
ds = ConfluenceDataSource(client)
coros = [
ds.get_custom_content_attachments(
id=i,
filename=f"file_{i}.dat",
limit=i % 5 + 1
)
for i in range(100, 120)
]
results = await asyncio.gather(*coros)
for i, r in zip(range(100, 120), results):
pass

@pytest.mark.asyncio

#------------------------------------------------
import asyncio # used to run async functions

---- Function under test (EXACT COPY) ----

from typing import Any, Dict, Optional, Union

import pytest # used for our unit tests
from app.sources.external.confluence.confluence import ConfluenceDataSource

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

These stubs allow us to test the async function deterministically.

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
self.data = getattr(response, 'data', None)
self.status_code = getattr(response, 'status_code', 200)
self.headers = getattr(response, 'headers', {})

class DummyAsyncResponse:
def init(self, data, status_code=200, headers=None):
self.data = data
self.status_code = status_code
self.headers = headers or {}

class DummyAsyncClient:
def init(self, base_url):
self._base_url = base_url
self.last_request = None

def get_base_url(self):
    return self._base_url

async def execute(self, request):
    # Save last request for inspection
    self.last_request = request
    # Simulate a response based on request parameters for testing
    resp_data = {
        'method': request.method,
        'url': request.url,
        'headers': request.headers,
        'path_params': request.path_params,
        'query_params': request.query_params,
        'body': request.body,
    }
    return HTTPResponse(DummyAsyncResponse(resp_data, status_code=200))

class DummyConfluenceClient:
def init(self, base_url):
self._client = DummyAsyncClient(base_url)

def get_client(self):
    return self._client

from app.sources.external.confluence.confluence import ConfluenceDataSource

---- Unit Tests ----

Basic Test Cases

@pytest.mark.asyncio
async def test_get_custom_content_attachments_basic():
"""Test basic async/await behavior and expected output structure"""
client = DummyConfluenceClient("https://example.atlassian.net")
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_attachments(id=123)
# Check that the returned data contains expected keys
for key in ['method', 'url', 'headers', 'path_params', 'query_params', 'body']:
pass

@pytest.mark.asyncio
async def test_get_custom_content_attachments_with_all_params():
"""Test with all optional parameters provided"""
client = DummyConfluenceClient("https://example.atlassian.net/")
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_attachments(
id=456,
sort={'by': 'date', 'order': 'desc'},
cursor="abc123",
status=['current', 'archived'],
mediaType="application/pdf",
filename="report.pdf",
limit=10,
headers={'Authorization': 'Bearer token123'}
)
# Check that query params are stringified
qp = resp.data['query_params']

@pytest.mark.asyncio
async def test_get_custom_content_attachments_empty_status():
"""Test with empty status list"""
client = DummyConfluenceClient("https://base.url")
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_attachments(id=1, status=[])

@pytest.mark.asyncio
async def test_get_custom_content_attachments_none_headers():
"""Test with headers=None"""
client = DummyConfluenceClient("https://base.url")
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_attachments(id=2, headers=None)

Edge Test Cases

@pytest.mark.asyncio
async def test_get_custom_content_attachments_invalid_client_raises():
"""Test 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_attachments_missing_get_base_url():
"""Test ValueError is raised if client does not have get_base_url method"""
class ClientNoBaseUrl:
def get_client(self):
class NoBaseUrl:
pass
return NoBaseUrl()
with pytest.raises(ValueError, match='HTTP client does not have get_base_url method'):
ConfluenceDataSource(ClientNoBaseUrl())

@pytest.mark.asyncio
async def test_get_custom_content_attachments_concurrent_execution():
"""Test concurrent execution with asyncio.gather"""
client = DummyConfluenceClient("https://base.url")
ds = ConfluenceDataSource(client)
# Run multiple requests concurrently
ids = [10, 20, 30, 40, 50]
tasks = [ds.get_custom_content_attachments(id=i) for i in ids]
results = await asyncio.gather(*tasks)
for i, resp in zip(ids, results):
pass

@pytest.mark.asyncio
async def test_get_custom_content_attachments_edge_types_in_query():
"""Test edge types for query parameters (bools, sets, tuples)"""
client = DummyConfluenceClient("https://base.url")
ds = ConfluenceDataSource(client)
resp = await ds.get_custom_content_attachments(
id=99,
status=[True, False, 'active'],
sort={'by': 'name'},
limit=0
)

Large Scale Test Cases

@pytest.mark.asyncio
async def test_get_custom_content_attachments_large_scale_concurrency():
"""Test function under moderate concurrent load"""
client = DummyConfluenceClient("https://base.url")
ds = ConfluenceDataSource(client)
# 50 concurrent requests
ids = list(range(100, 150))
tasks = [ds.get_custom_content_attachments(id=i, filename=f"file_{i}.txt") for i in ids]
results = await asyncio.gather(*tasks)
for i, resp in zip(ids, results):
pass

@pytest.mark.asyncio
async def test_get_custom_content_attachments_large_query_dict():
"""Test with a large sort dict in query params"""
client = DummyConfluenceClient("https://base.url")
ds = ConfluenceDataSource(client)
sort_dict = {f"field_{i}": f"value_{i}" for i in range(50)}
resp = await ds.get_custom_content_attachments(id=1, sort=sort_dict)

Throughput Test Cases

@pytest.mark.asyncio
async def test_get_custom_content_attachments_throughput_small_load():
"""Throughput test: small load, 5 requests"""
client = DummyConfluenceClient("https://base.url")
ds = ConfluenceDataSource(client)
ids = [1, 2, 3, 4, 5]
tasks = [ds.get_custom_content_attachments(id=i) for i in ids]
results = await asyncio.gather(*tasks)
for i, resp in zip(ids, results):
pass

@pytest.mark.asyncio

async def test_get_custom_content_attachments_throughput_varied_parameters():
"""Throughput test: varied parameters, 20 requests"""
client = DummyConfluenceClient("https://base.url")
ds = ConfluenceDataSource(client)
tasks = []
for i in range(20):
params = dict(
id=i,
sort={'by': 'name', 'order': 'asc'} if i % 2 == 0 else None,
status=['active', 'archived'] if i % 3 == 0 else None,
mediaType="application/pdf" if i % 4 == 0 else None,
filename=f"doc_{i}.pdf" if i % 5 == 0 else None,
limit=i if i % 6 == 0 else None
)
tasks.append(ds.get_custom_content_attachments(**params))
results = await asyncio.gather(*tasks)
for i, resp in enumerate(results):
# Check that optional fields are present or absent as expected
if i % 2 == 0:
pass
if i % 3 == 0:
pass
if i % 4 == 0:
pass
if i % 5 == 0:
pass
if i % 6 == 0:
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_attachments-mhve2pe4 and push.