⚡️ Speed up method BookStackDataSource.get_content_permissions by 13%

[codeflash-ai]

📄 13% (0.13x) speedup for BookStackDataSource.get_content_permissions in backend/python/app/sources/external/bookstack/bookstack.py

⏱️ Runtime : 1.92 milliseconds → 1.69 milliseconds (best of 283 runs)

📝 Explanation and details

The optimization achieves a 13% runtime improvement through two key micro-optimizations in the get_content_permissions method:

1. String Formatting Optimization

• Original: url = self.base_url + "/api/content-permissions/{content_type}/{content_id}".format(content_type=content_type, content_id=content_id)
• Optimized: url = f"{self.base_url}/api/content-permissions/{content_type}/{content_id}"

The f-string approach eliminates the overhead of .format() method calls and dictionary lookups for variable substitution. Line profiler shows this reduced execution time from 428,354ns to 218,124ns (49% improvement on this line).

2. Header Copy Elimination

• Original: headers = dict(self.http.headers)
• Optimized: headers = self.http.headers

Since headers are only read (not mutated) within the request, the unnecessary dict copy is removed. This reduces execution time from 188,413ns to 127,697ns (32% improvement on this line).

Performance Impact: These optimizations are particularly effective for high-frequency API calls, as evidenced by the test cases with concurrent requests (10-100 operations). The 13% runtime improvement comes from eliminating repetitive overhead in two of the most frequently executed lines during request construction.

Test Case Effectiveness: The optimizations show consistent benefits across all test scenarios, from basic single calls to throughput tests with 100 concurrent requests, making this especially valuable for applications making frequent BookStack API calls.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 563 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 (EXACTLY as provided)
from typing import Dict, Union

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

# Mocks for HTTPRequest, HTTPResponse, BookStackResponse, HTTPClient, BookStackRESTClientViaToken, BookStackClient

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 BookStackResponse:
        def __init__(self, success, data=None, error=None):
            self.success = success
            self.data = data
            self.error = error

# Minimal mock HTTPClient that simulates async HTTP execution
class MockHTTPClient:
    def __init__(self, headers=None, response_data=None, raise_exc=None):
        self.headers = headers or {"Authorization": "Token testid:testsecret"}
        self._response_data = response_data
        self._raise_exc = raise_exc

    def get_base_url(self):
        return "https://mock.bookstack"

    async def execute(self, request, **kwargs):
        if self._raise_exc:
            raise self._raise_exc
        # Simulate a response with the given data
        return HTTPResponse(self._response_data)

# Minimal BookStackClient for dependency injection
class BookStackClient:
    def __init__(self, http):
        self._http = http

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

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

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_get_content_permissions_returns_success_and_data():
    """Test normal case: valid content_type and content_id returns success with expected data."""
    expected_data = {"permissions": ["view", "edit"], "id": 123}
    http_client = MockHTTPClient(response_data=expected_data)
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    result = await datasource.get_content_permissions("page", 123)

@pytest.mark.asyncio
async def test_get_content_permissions_basic_async_await():
    """Test that function is awaitable and returns BookStackResponse."""
    http_client = MockHTTPClient(response_data={"permissions": []})
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    codeflash_output = datasource.get_content_permissions("book", 1); coro = codeflash_output
    result = await coro

@pytest.mark.asyncio
async def test_get_content_permissions_url_and_headers():
    """Test that the constructed URL and headers are correct."""
    captured = {}
    class CaptureHTTPClient(MockHTTPClient):
        async def execute(self, request, **kwargs):
            captured['url'] = request.url
            captured['headers'] = request.headers
            return HTTPResponse({"ok": True})
    http_client = CaptureHTTPClient()
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    await datasource.get_content_permissions("chapter", 42)

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_get_content_permissions_invalid_content_type():
    """Test with an invalid content_type (should still call execute, server may reject)."""
    http_client = MockHTTPClient(response_data={"error": "Invalid content type"})
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    result = await datasource.get_content_permissions("invalid_type", 99)

@pytest.mark.asyncio
async def test_get_content_permissions_negative_content_id():
    """Test with a negative content_id (should still call execute)."""
    http_client = MockHTTPClient(response_data={"permissions": [], "id": -1})
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    result = await datasource.get_content_permissions("page", -1)

@pytest.mark.asyncio
async def test_get_content_permissions_execute_raises_exception():
    """Test that if the HTTP client raises an exception, error is returned and success is False."""
    http_client = MockHTTPClient(raise_exc=RuntimeError("network error"))
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    result = await datasource.get_content_permissions("book", 1)

@pytest.mark.asyncio
async def test_get_content_permissions_http_client_not_initialized():
    """Test that constructor raises ValueError if HTTP client is None."""
    class NullBookStackClient:
        def get_client(self):
            return None
    with pytest.raises(ValueError, match="HTTP client is not initialized"):
        BookStackDataSource(NullBookStackClient())

@pytest.mark.asyncio
async def test_get_content_permissions_http_client_missing_get_base_url():
    """Test that constructor raises ValueError if HTTP client lacks get_base_url."""
    class NoBaseURLClient:
        headers = {}
    class DummyBookStackClient:
        def get_client(self):
            return NoBaseURLClient()
    with pytest.raises(ValueError, match="HTTP client does not have get_base_url method"):
        BookStackDataSource(DummyBookStackClient())

@pytest.mark.asyncio
async def test_get_content_permissions_concurrent_calls():
    """Test concurrent execution of get_content_permissions."""
    http_client = MockHTTPClient(response_data={"permissions": ["view"]})
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    # Run 5 concurrent calls
    results = await asyncio.gather(
        *[datasource.get_content_permissions("page", i) for i in range(5)]
    )
    for result in results:
        pass

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_get_content_permissions_many_concurrent_calls():
    """Test the function with a large number of concurrent calls (up to 100)."""
    http_client = MockHTTPClient(response_data={"permissions": ["view"]})
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    tasks = [
        datasource.get_content_permissions("page", i)
        for i in range(100)
    ]
    results = await asyncio.gather(*tasks)
    for idx, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_get_content_permissions_large_content_id():
    """Test with a very large content_id value."""
    http_client = MockHTTPClient(response_data={"permissions": ["edit"], "id": 999999999})
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    result = await datasource.get_content_permissions("book", 999999999)

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_get_content_permissions_throughput_small_load():
    """Throughput test: small batch of requests."""
    http_client = MockHTTPClient(response_data={"permissions": ["view"]})
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    tasks = [datasource.get_content_permissions("page", i) for i in range(10)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_get_content_permissions_throughput_medium_load():
    """Throughput test: medium batch of requests."""
    http_client = MockHTTPClient(response_data={"permissions": ["edit"]})
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    tasks = [datasource.get_content_permissions("chapter", i) for i in range(50)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_get_content_permissions_throughput_high_volume():
    """Throughput test: high volume of requests (within safe limits)."""
    http_client = MockHTTPClient(response_data={"permissions": ["view", "edit"]})
    client = BookStackClient(http_client)
    datasource = BookStackDataSource(client)
    tasks = [datasource.get_content_permissions("bookshelf", i) for i in range(100)]
    results = await asyncio.gather(*tasks)
    for result 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.
#------------------------------------------------
import asyncio  # used to run async functions
# --- The function under test (EXACT COPY, DO NOT MODIFY) ---
from typing import Dict, Union

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

# --- Minimal stubs for dependencies required by get_content_permissions ---

# HTTPRequest stub
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 = {}  # not used in get_content_permissions

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

# Minimal httpx.Response stub for json()
class DummyResponse:
    def __init__(self, data):
        self._data = data

    def json(self):
        return self._data

# Minimal Async HTTPClient stub for testing
class DummyHTTPClient:
    def __init__(self, headers=None, should_raise=False, response_data=None):
        self.headers = headers or {"Authorization": "Token dummy:dummy"}
        self._should_raise = should_raise
        self._response_data = response_data or {"permissions": "read"}
        self._executed_requests = []

    def get_base_url(self):
        return "https://example.com"

    async def execute(self, request, **kwargs):
        self._executed_requests.append(request)
        if self._should_raise:
            raise RuntimeError("Simulated HTTP error")
        # Simulate a response object
        return HTTPResponse(DummyResponse(self._response_data))

# BookStackClient stub
class BookStackClient:
    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 for async get_content_permissions ---

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_get_content_permissions_basic_success():
    """Test basic successful permission fetch for a book"""
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "read"}))
    datasource = BookStackDataSource(client)
    result = await datasource.get_content_permissions("book", 123)

@pytest.mark.asyncio
async def test_get_content_permissions_basic_different_content_types():
    """Test permissions fetch for all supported content types"""
    for content_type in ["page", "book", "chapter", "bookshelf"]:
        client = BookStackClient(DummyHTTPClient(response_data={"permissions": f"{content_type}-perm"}))
        datasource = BookStackDataSource(client)
        result = await datasource.get_content_permissions(content_type, 1)

@pytest.mark.asyncio
async def test_get_content_permissions_basic_async_await():
    """Test that the function returns a coroutine and can be awaited"""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    codeflash_output = datasource.get_content_permissions("page", 1); coro = codeflash_output
    result = await coro

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_get_content_permissions_edge_invalid_content_type():
    """Test with an invalid content type (should still succeed, as function does not validate type)"""
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "unknown"}))
    datasource = BookStackDataSource(client)
    result = await datasource.get_content_permissions("notarealtype", 42)

@pytest.mark.asyncio
async def test_get_content_permissions_edge_negative_content_id():
    """Test with a negative content_id (should still succeed, as function does not validate id)"""
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "negative"}))
    datasource = BookStackDataSource(client)
    result = await datasource.get_content_permissions("book", -1)

@pytest.mark.asyncio
async def test_get_content_permissions_edge_zero_content_id():
    """Test with zero as content_id (should succeed, as function does not validate id)"""
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "zero"}))
    datasource = BookStackDataSource(client)
    result = await datasource.get_content_permissions("page", 0)

@pytest.mark.asyncio
async def test_get_content_permissions_edge_http_exception():
    """Test that function handles HTTP client exceptions gracefully"""
    client = BookStackClient(DummyHTTPClient(should_raise=True))
    datasource = BookStackDataSource(client)
    result = await datasource.get_content_permissions("book", 123)

@pytest.mark.asyncio
async def test_get_content_permissions_edge_concurrent_execution():
    """Test concurrent execution of multiple permission requests"""
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "read"}))
    datasource = BookStackDataSource(client)
    coros = [
        datasource.get_content_permissions("page", i)
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_get_content_permissions_large_scale_concurrent():
    """Test large scale concurrent execution with unique content_ids"""
    N = 50  # Keep under 1000 for speed
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "bulk"}))
    datasource = BookStackDataSource(client)
    coros = [
        datasource.get_content_permissions("book", i)
        for i in range(N)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass

@pytest.mark.asyncio
async def test_get_content_permissions_large_scale_varied_types():
    """Test large scale concurrent execution with varied content types"""
    types = ["page", "book", "chapter", "bookshelf"]
    N = 20
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "multi"}))
    datasource = BookStackDataSource(client)
    coros = [
        datasource.get_content_permissions(types[i % len(types)], i)
        for i in range(N)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_get_content_permissions_throughput_small_load():
    """Throughput test: small load of 10 concurrent requests"""
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "small"}))
    datasource = BookStackDataSource(client)
    coros = [
        datasource.get_content_permissions("page", i)
        for i in range(10)
    ]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_get_content_permissions_throughput_medium_load():
    """Throughput test: medium load of 50 concurrent requests"""
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "medium"}))
    datasource = BookStackDataSource(client)
    coros = [
        datasource.get_content_permissions("book", i)
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_get_content_permissions_throughput_high_volume():
    """Throughput test: high volume of 100 concurrent requests"""
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "high"}))
    datasource = BookStackDataSource(client)
    coros = [
        datasource.get_content_permissions("chapter", i)
        for i in range(100)
    ]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_get_content_permissions_throughput_sustained_pattern():
    """Throughput test: sustained execution pattern (sequential then concurrent)"""
    client = BookStackClient(DummyHTTPClient(response_data={"permissions": "sustained"}))
    datasource = BookStackDataSource(client)
    # Sequential
    for i in range(5):
        result = await datasource.get_content_permissions("bookshelf", i)
    # Concurrent
    coros = [datasource.get_content_permissions("bookshelf", i) for i in range(5, 15)]
    results = await asyncio.gather(*coros)
# 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.get_content_permissions-mhbogym8 and push.