⚡️ Speed up method BookStackDataSource.export_chapter_pdf by 7%

[codeflash-ai]

📄 7% (0.07x) speedup for BookStackDataSource.export_chapter_pdf in backend/python/app/sources/external/bookstack/bookstack.py

⏱️ Runtime : 1.24 milliseconds → 1.15 milliseconds (best of 306 runs)

📝 Explanation and details

The optimization achieves a 7% runtime improvement through three key micro-optimizations:

1. Conditional Header Merging in HTTPClient

• Original: Always creates new dict with {**self.headers, **request.headers} even when request.headers is empty
• Optimized: Only merges when request.headers has content, otherwise reuses self.headers directly
• Impact: Eliminates unnecessary dictionary creation and copying operations in the common case where requests have no custom headers

2. Efficient kwargs Construction

• Original: Uses **kwargs unpacking in dict literal creation
• Optimized: Conditionally calls request_kwargs.update(kwargs) only when kwargs exist
• Impact: Avoids unpacking overhead when no additional arguments are passed

3. Direct Header Reference in BookStackDataSource

• Original: Creates new dict copy with dict(self.http.headers)
• Optimized: Uses self.http.headers directly (safe since headers are immutable after initialization)
• Impact: Eliminates dictionary copying overhead entirely

4. F-string URL Formatting

• Original: String concatenation with .format() method call
• Optimized: Direct f-string interpolation f"{self.base_url}/api/chapters/{id}/export/pdf"
• Impact: Slightly faster string formatting

Performance Profile Analysis:

• Line profiler shows the most significant gains in URL construction (9.0% → 5.4% of total time) and header handling (4.4% → 3.3%)
• The optimizations are most effective for high-frequency request scenarios where these micro-optimizations compound across many calls
• Test cases with concurrent requests (50-100 operations) benefit most from reduced allocation overhead

These optimizations target the hot path of HTTP request preparation, making them particularly valuable for applications making many API calls with consistent header patterns.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
# --- Function under test (EXACT COPY) ---
from typing import Dict, Union

# Import classes from the code under test
import httpx
import pytest  # used for our unit tests
from app.sources.external.bookstack.bookstack import BookStackDataSource


# --- Minimal stubs for dependencies (for isolated testing) ---
class HTTPRequest:
    def __init__(self, method, url, headers, query_params, body):
        self.method = method
        self.url = url
        self.headers = headers
        self.query_params = query_params
        self.body = body
        self.path_params = {}

class BookStackResponse:
        def __init__(self, success, data=None, error=None):
            self.success = success
            self.data = data
            self.error = error

# --- Mocks for HTTP client and BookStack client ---

class DummyAsyncResponse:
    def __init__(self, json_data):
        self._json_data = json_data

    def json(self):
        return self._json_data

class DummyHTTPClient:
    """A dummy async HTTP client for simulating real HTTP calls."""
    def __init__(self):
        self.headers = {"Authorization": "Token testtoken"}
        self._responses = {}
        self._should_raise = {}

    def set_response(self, url, json_data):
        self._responses[url] = json_data

    def set_exception(self, url, exc):
        self._should_raise[url] = exc

    async def execute(self, request, **kwargs):
        url = request.url
        if url in self._should_raise:
            raise self._should_raise[url]
        return HTTPResponse(DummyAsyncResponse(self._responses.get(url, {"dummy": True})))

    def get_base_url(self):
        return "http://localhost"

class DummyBookStackClient:
    """A dummy BookStackClient for dependency injection."""
    def __init__(self, http_client):
        self._http_client = http_client

    def get_client(self):
        return self._http_client
from app.sources.external.bookstack.bookstack import BookStackDataSource

# ------------------- UNIT TESTS -------------------

# ---------- BASIC TEST CASES ----------

@pytest.mark.asyncio
async def test_export_chapter_pdf_basic_success():
    """Test basic successful export of a chapter as PDF."""
    dummy_http = DummyHTTPClient()
    chapter_id = 123
    expected_json = {"chapter_id": chapter_id, "pdf_url": "http://localhost/api/chapters/123/export/pdf"}
    url = f"http://localhost/api/chapters/{chapter_id}/export/pdf"
    dummy_http.set_response(url, expected_json)
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    # Await the async function and check the result
    result = await datasource.export_chapter_pdf(chapter_id)

@pytest.mark.asyncio
async def test_export_chapter_pdf_returns_bookstackresponse():
    """Test that the function returns a BookStackResponse object."""
    dummy_http = DummyHTTPClient()
    chapter_id = 1
    url = f"http://localhost/api/chapters/{chapter_id}/export/pdf"
    dummy_http.set_response(url, {"foo": "bar"})
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    result = await datasource.export_chapter_pdf(chapter_id)

@pytest.mark.asyncio
async def test_export_chapter_pdf_basic_async_behavior():
    """Test that the function is properly async and can be awaited."""
    dummy_http = DummyHTTPClient()
    chapter_id = 42
    url = f"http://localhost/api/chapters/{chapter_id}/export/pdf"
    dummy_http.set_response(url, {"chapter_id": chapter_id})
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    # Call without await to check coroutine
    codeflash_output = datasource.export_chapter_pdf(chapter_id); coro = codeflash_output
    result = await coro

# ---------- EDGE TEST CASES ----------

@pytest.mark.asyncio
async def test_export_chapter_pdf_http_exception():
    """Test that exceptions in the HTTP client are caught and returned as error."""
    dummy_http = DummyHTTPClient()
    chapter_id = 999
    url = f"http://localhost/api/chapters/{chapter_id}/export/pdf"
    dummy_http.set_exception(url, RuntimeError("HTTP error occurred"))
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    result = await datasource.export_chapter_pdf(chapter_id)

@pytest.mark.asyncio
async def test_export_chapter_pdf_invalid_http_client():
    """Test that initializing with a client that returns None raises ValueError."""
    class BadClient:
        def get_client(self): return None
    with pytest.raises(ValueError):
        BookStackDataSource(BadClient())

@pytest.mark.asyncio
async def test_export_chapter_pdf_http_client_missing_base_url():
    """Test that client missing get_base_url raises ValueError."""
    class BadHTTP:
        headers = {}
    class BadClient:
        def get_client(self): return BadHTTP()
    with pytest.raises(ValueError):
        BookStackDataSource(BadClient())

@pytest.mark.asyncio
async def test_export_chapter_pdf_concurrent_calls():
    """Test concurrent export_chapter_pdf calls with different IDs."""
    dummy_http = DummyHTTPClient()
    ids = [10, 20, 30]
    for cid in ids:
        url = f"http://localhost/api/chapters/{cid}/export/pdf"
        dummy_http.set_response(url, {"chapter_id": cid, "pdf_url": url})
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    # Run concurrent requests
    results = await asyncio.gather(
        *(datasource.export_chapter_pdf(cid) for cid in ids)
    )
    for i, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_export_chapter_pdf_edge_case_zero_id():
    """Test export with chapter id=0 (edge case)."""
    dummy_http = DummyHTTPClient()
    chapter_id = 0
    url = f"http://localhost/api/chapters/{chapter_id}/export/pdf"
    dummy_http.set_response(url, {"chapter_id": chapter_id})
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    result = await datasource.export_chapter_pdf(chapter_id)

@pytest.mark.asyncio
async def test_export_chapter_pdf_edge_case_negative_id():
    """Test export with negative chapter id (edge case)."""
    dummy_http = DummyHTTPClient()
    chapter_id = -5
    url = f"http://localhost/api/chapters/{chapter_id}/export/pdf"
    dummy_http.set_response(url, {"chapter_id": chapter_id})
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    result = await datasource.export_chapter_pdf(chapter_id)

# ---------- LARGE SCALE TEST CASES ----------

@pytest.mark.asyncio
async def test_export_chapter_pdf_many_concurrent_requests():
    """Test many concurrent export_chapter_pdf calls to ensure scalability."""
    dummy_http = DummyHTTPClient()
    ids = list(range(1, 51))  # 50 concurrent requests (well under 1000)
    for cid in ids:
        url = f"http://localhost/api/chapters/{cid}/export/pdf"
        dummy_http.set_response(url, {"chapter_id": cid})
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    results = await asyncio.gather(
        *(datasource.export_chapter_pdf(cid) for cid in ids)
    )
    for i, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_export_chapter_pdf_large_id():
    """Test export with a very large chapter id."""
    dummy_http = DummyHTTPClient()
    chapter_id = 987654321
    url = f"http://localhost/api/chapters/{chapter_id}/export/pdf"
    dummy_http.set_response(url, {"chapter_id": chapter_id})
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    result = await datasource.export_chapter_pdf(chapter_id)

# ---------- THROUGHPUT TEST CASES ----------

@pytest.mark.asyncio
async def test_export_chapter_pdf_throughput_small_load():
    """Throughput: Test function with a small batch of requests."""
    dummy_http = DummyHTTPClient()
    ids = [1, 2, 3, 4, 5]
    for cid in ids:
        url = f"http://localhost/api/chapters/{cid}/export/pdf"
        dummy_http.set_response(url, {"chapter_id": cid})
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    results = await asyncio.gather(
        *(datasource.export_chapter_pdf(cid) for cid in ids)
    )

@pytest.mark.asyncio
async def test_export_chapter_pdf_throughput_medium_load():
    """Throughput: Test function with a medium batch of requests."""
    dummy_http = DummyHTTPClient()
    ids = list(range(100, 130))  # 30 requests
    for cid in ids:
        url = f"http://localhost/api/chapters/{cid}/export/pdf"
        dummy_http.set_response(url, {"chapter_id": cid})
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    results = await asyncio.gather(
        *(datasource.export_chapter_pdf(cid) for cid in ids)
    )

@pytest.mark.asyncio
async def test_export_chapter_pdf_throughput_high_volume():
    """Throughput: Test function with a high but bounded volume of requests."""
    dummy_http = DummyHTTPClient()
    ids = list(range(500, 600))  # 100 requests (well under 1000)
    for cid in ids:
        url = f"http://localhost/api/chapters/{cid}/export/pdf"
        dummy_http.set_response(url, {"chapter_id": cid})
    datasource = BookStackDataSource(DummyBookStackClient(dummy_http))

    results = await asyncio.gather(
        *(datasource.export_chapter_pdf(cid) for cid in ids)
    )
# 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 required classes for constructing test environments
from typing import Any

import pytest  # used for our unit tests
from app.sources.external.bookstack.bookstack import BookStackDataSource

# --- Minimal stubs for required classes ---

class HTTPRequest:
    def __init__(self, method, url, headers=None, query_params=None, body=None, path_params=None):
        self.method = method
        self.url = url
        self.headers = headers or {}
        self.query_params = query_params or {}
        self.body = body
        self.path_params = path_params or {}

class DummyResponse:
    def __init__(self, content):
        self._content = content

    def json(self):
        return self._content

class DummyAsyncClient:
    # Simulates httpx.AsyncClient
    async def request(self, method, url, **kwargs):
        # Simulate a response object
        # For edge cases, use the url or headers to decide response
        if "fail" in url:
            raise Exception("Simulated HTTP error")
        if "large" in url:
            # Simulate a large response
            return DummyResponse({"pdf": "X" * 10000, "chapter_id": int(url.split('/')[-3])})
        # Default: echo back the chapter id
        try:
            chapter_id = int(url.split('/')[-3])
        except Exception:
            chapter_id = None
        return DummyResponse({"pdf": "PDF_DATA", "chapter_id": chapter_id})

class DummyHTTPClient:
    # Simulates HTTPClient
    def __init__(self):
        self.headers = {"Authorization": "Token TEST_TOKEN"}
        self.base_url = "http://testserver"
    async def execute(self, request, **kwargs):
        # Simulate response based on request
        return HTTPResponse(await DummyAsyncClient().request(request.method, request.url, **kwargs))
    def get_base_url(self):
        return self.base_url

class BookStackResponse:
    def __init__(self, success: bool, data: Any = None, error: Any = None):
        self.success = success
        self.data = data
        self.error = error
from app.sources.external.bookstack.bookstack import BookStackDataSource


# --- Minimal BookStackClient stub ---
class BookStackClient:
    def __init__(self, http_client):
        self.http_client = http_client
    def get_client(self):
        return self.http_client

# --- Test Fixtures ---

@pytest.fixture
def datasource():
    # Returns a BookStackDataSource with a dummy client
    client = BookStackClient(DummyHTTPClient())
    return BookStackDataSource(client)

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

@pytest.mark.asyncio
async def test_export_chapter_pdf_basic_success(datasource):
    """Test basic successful PDF export for a valid chapter id."""
    result = await datasource.export_chapter_pdf(123)

@pytest.mark.asyncio
async def test_export_chapter_pdf_basic_zero_id(datasource):
    """Test PDF export with chapter id zero."""
    result = await datasource.export_chapter_pdf(0)

@pytest.mark.asyncio
async def test_export_chapter_pdf_basic_negative_id(datasource):
    """Test PDF export with negative chapter id."""
    result = await datasource.export_chapter_pdf(-5)

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

@pytest.mark.asyncio
async def test_export_chapter_pdf_edge_invalid_url(datasource):
    """Test PDF export with a chapter id that triggers a simulated failure."""
    # Patch base_url to include 'fail' to simulate error
    datasource.base_url = "http://testserver/fail"
    result = await datasource.export_chapter_pdf(999)

@pytest.mark.asyncio
async def test_export_chapter_pdf_edge_missing_http_client():
    """Test error raised if HTTP client is not initialized."""
    class DummyBadClient:
        def get_client(self):
            return None
    with pytest.raises(ValueError) as excinfo:
        BookStackDataSource(DummyBadClient())

@pytest.mark.asyncio
async def test_export_chapter_pdf_edge_missing_base_url():
    """Test error raised if HTTP client lacks get_base_url."""
    class DummyBadHTTPClient:
        pass
    class DummyBadClient:
        def get_client(self):
            return DummyBadHTTPClient()
    with pytest.raises(ValueError) as excinfo:
        BookStackDataSource(DummyBadClient())

@pytest.mark.asyncio
async def test_export_chapter_pdf_edge_concurrent_requests(datasource):
    """Test concurrent PDF exports for different chapter ids."""
    ids = [1, 2, 3, 4, 5]
    # Run all requests concurrently
    results = await asyncio.gather(*(datasource.export_chapter_pdf(i) for i in ids))
    # All should succeed and have correct chapter ids
    for idx, result in zip(ids, results):
        pass

@pytest.mark.asyncio
async def test_export_chapter_pdf_edge_headers_are_copied(datasource):
    """Test that headers are copied from HTTP client."""
    result = await datasource.export_chapter_pdf(42)

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

@pytest.mark.asyncio
async def test_export_chapter_pdf_large_scale_many_concurrent(datasource):
    """Test many concurrent PDF exports for scalability."""
    ids = list(range(10, 60))  # 50 concurrent requests
    results = await asyncio.gather(*(datasource.export_chapter_pdf(i) for i in ids))
    # All should succeed and have correct chapter ids
    for idx, result in zip(ids, results):
        pass

@pytest.mark.asyncio
async def test_export_chapter_pdf_large_scale_large_response(datasource):
    """Test export with a simulated large response."""
    # Patch base_url to include 'large' to trigger large response
    datasource.base_url = "http://testserver/large"
    result = await datasource.export_chapter_pdf(555)

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

@pytest.mark.asyncio
async def test_export_chapter_pdf_throughput_small_load(datasource):
    """Throughput: Test small batch of PDF exports for quick completion."""
    ids = [101, 102, 103]
    results = await asyncio.gather(*(datasource.export_chapter_pdf(i) for i in ids))
    for idx, result in zip(ids, results):
        pass

@pytest.mark.asyncio
async def test_export_chapter_pdf_throughput_medium_load(datasource):
    """Throughput: Test medium batch of PDF exports."""
    ids = list(range(200, 220))  # 20 requests
    results = await asyncio.gather(*(datasource.export_chapter_pdf(i) for i in ids))
    for idx, result in zip(ids, results):
        pass

@pytest.mark.asyncio
async def test_export_chapter_pdf_throughput_high_volume(datasource):
    """Throughput: Test higher volume batch of PDF exports."""
    ids = list(range(300, 350))  # 50 requests
    results = await asyncio.gather(*(datasource.export_chapter_pdf(i) for i in ids))
    for idx, result in zip(ids, results):
        pass

@pytest.mark.asyncio
async def test_export_chapter_pdf_throughput_mixed_load(datasource):
    """Throughput: Test mixed valid and error-inducing requests."""
    # Some ids will trigger error by patching base_url
    ids = [401, 402, 403, 404, 405]
    original_base_url = datasource.base_url
    # Patch for some ids
    def get_url(id):
        if id % 2 == 0:
            return original_base_url
        else:
            return original_base_url + "/fail"
    results = await asyncio.gather(
        *(BookStackDataSource(BookStackClient(DummyHTTPClient())).export_chapter_pdf(i)
          if i % 2 == 0 else
          BookStackDataSource(BookStackClient(DummyHTTPClient())).export_chapter_pdf(i)
          for i in ids)
    )
    # Odd ids should fail, even ids should succeed
    for idx, result in zip(ids, results):
        if idx % 2 == 0:
            pass
        else:
            pass
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
from app.sources.external.bookstack.bookstack import BookStackDataSource

To edit these changes git checkout codeflash/optimize-BookStackDataSource.export_chapter_pdf-mhbjh8t3 and push.