⚡️ Speed up method BookStackDataSource.update_content_permissions by 12%

[codeflash-ai]

📄 12% (0.12x) speedup for BookStackDataSource.update_content_permissions in backend/python/app/sources/external/bookstack/bookstack.py

⏱️ Runtime : 1.44 milliseconds → 1.28 milliseconds (best of 291 runs)

📝 Explanation and details

The optimized code achieves an 11% runtime speedup through two key optimizations that target the most time-consuming operations identified in the line profiler:

1. URL Construction Optimization (Major Impact)

• Original: url = self.base_url + "/api/content-permissions/{content_type}/{content_id}".format(content_type=content_type, content_id=content_id) took 287,299ns (7.5% of total time)
• Optimized: url = f"{self.base_url}/api/content-permissions/{content_type}/{content_id}" takes only 158,076ns (4.5% of total time)
• Why faster: F-strings are compiled into more efficient bytecode than .format() method calls, eliminating the overhead of method dispatch and dictionary lookups for named parameters

2. Header Dictionary Copy Optimization

• Original: headers = dict(self.http.headers) creates a new dict by calling the dict constructor
• Optimized: headers = self.http.headers.copy() uses the built-in copy method which is more efficient for existing dictionaries
• Why faster: .copy() is implemented in C and avoids the overhead of the dict constructor and iterator protocol

3. Conditional URL Formatting (HTTPClient)

• Added a check to skip .format() when path_params is empty, avoiding unnecessary string operations
• This optimization primarily benefits cases where URL templates don't need parameter substitution

Performance Profile: These optimizations are most effective for high-throughput scenarios where the function is called repeatedly (as shown in the large-scale and throughput test cases), since they reduce per-call overhead in string operations and dictionary manipulations. The 0.3% throughput improvement demonstrates consistent gains across concurrent operations, making this optimization valuable for API-heavy workloads.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
# The function under test (EXACT copy, DO NOT MODIFY)
from typing import Dict, List, Optional, Union

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

# Minimal stubs for required classes/objects for the test suite.
# These are necessary for the function to run and be tested in isolation.

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

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

class DummyHTTPResponse:
    def __init__(self, json_data):
        self._json_data = json_data
    def json(self):
        return self._json_data

class DummyHTTPClient:
    def __init__(self, should_fail=False, fail_with=None):
        # Simulate headers and base_url needed by BookStackDataSource
        self.headers = {"Authorization": "Token dummy"}
        self._base_url = "http://localhost"
        self.should_fail = should_fail
        self.fail_with = fail_with
        self.executed_requests = []

    def get_base_url(self):
        return self._base_url

    async def execute(self, request, **kwargs):
        self.executed_requests.append(request)
        if self.should_fail:
            if self.fail_with:
                raise self.fail_with
            raise Exception("Simulated HTTP failure")
        # Simulate a normal response
        # For testing, echo back the body and url for validation
        return DummyHTTPResponse({
            "method": request.method,
            "url": request.url,
            "headers": request.headers,
            "query_params": request.query_params,
            "body": request.body,
        })

class BookStackClient:
    def __init__(self, client):
        self.client = client
    def get_client(self):
        return self.client
from app.sources.external.bookstack.bookstack import BookStackDataSource

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

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_update_content_permissions_basic_minimal():
    """Test minimal valid input - only required params."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    result = await datasource.update_content_permissions("page", 42)

@pytest.mark.asyncio
async def test_update_content_permissions_basic_full_params():
    """Test all optional parameters provided."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    role_permissions = [{"role_id": 1, "view": True}, {"role_id": 2, "edit": False}]
    fallback_permissions = {"view": True, "edit": False}
    result = await datasource.update_content_permissions(
        content_type="chapter",
        content_id=123,
        owner_id=99,
        role_permissions=role_permissions,
        fallback_permissions=fallback_permissions
    )

@pytest.mark.asyncio
async def test_update_content_permissions_basic_async_await_behavior():
    """Test that the returned coroutine must be awaited and returns correct type."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    codeflash_output = datasource.update_content_permissions("book", 7); coro = codeflash_output
    result = await coro

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_update_content_permissions_edge_invalid_content_type():
    """Test with an invalid content_type string."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    # Should not raise, but return success True (function does not validate content_type)
    result = await datasource.update_content_permissions("notarealtype", 1)

@pytest.mark.asyncio
async def test_update_content_permissions_edge_negative_content_id():
    """Test with a negative content_id."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    result = await datasource.update_content_permissions("page", -5)

@pytest.mark.asyncio
async def test_update_content_permissions_edge_empty_role_permissions():
    """Test with empty role_permissions list."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    result = await datasource.update_content_permissions("book", 10, role_permissions=[])

@pytest.mark.asyncio
async def test_update_content_permissions_edge_none_fallback_permissions():
    """Test with fallback_permissions explicitly set to None."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    result = await datasource.update_content_permissions("chapter", 11, fallback_permissions=None)

@pytest.mark.asyncio
async def test_update_content_permissions_edge_concurrent_execution():
    """Test concurrent execution of multiple requests."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    # Launch several concurrent requests with different content IDs
    coros = [
        datasource.update_content_permissions("page", i)
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)
    for idx, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_update_content_permissions_edge_exception_handling():
    """Test that exceptions in HTTP client are caught and returned as error."""
    client = BookStackClient(DummyHTTPClient(should_fail=True))
    datasource = BookStackDataSource(client)
    result = await datasource.update_content_permissions("book", 99)

@pytest.mark.asyncio
async def test_update_content_permissions_edge_custom_exception_message():
    """Test that a custom exception is correctly propagated in error field."""
    class CustomError(Exception):
        pass
    client = BookStackClient(DummyHTTPClient(should_fail=True, fail_with=CustomError("Custom fail")))
    datasource = BookStackDataSource(client)
    result = await datasource.update_content_permissions("book", 100)

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_update_content_permissions_large_scale_many_concurrent():
    """Test scalability with many concurrent requests (50)."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    coros = [
        datasource.update_content_permissions("shelf", i)
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_update_content_permissions_large_scale_varied_params():
    """Test large scale with varied combinations of parameters."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    coros = []
    for i in range(20):
        role_permissions = [{"role_id": i, "view": bool(i % 2)}]
        fallback_permissions = {"view": bool(i % 2), "edit": bool((i+1) % 2)}
        coros.append(datasource.update_content_permissions(
            "book", i, owner_id=i+100, role_permissions=role_permissions, fallback_permissions=fallback_permissions
        ))
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_update_content_permissions_throughput_small_load():
    """Throughput: Test with a small load of 5 requests."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    coros = [datasource.update_content_permissions("page", i) for i in range(5)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_update_content_permissions_throughput_medium_load():
    """Throughput: Test with a medium load of 20 requests."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    coros = [datasource.update_content_permissions("chapter", i) for i in range(20)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_update_content_permissions_throughput_high_volume():
    """Throughput: Test with a high volume of 100 requests."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    coros = [datasource.update_content_permissions("book", i) for i in range(100)]
    results = await asyncio.gather(*coros)
    # Optionally, check that each request had the correct URL
    for i, r in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_update_content_permissions_throughput_mixed_params():
    """Throughput: Test with mixed parameters under load."""
    client = BookStackClient(DummyHTTPClient())
    datasource = BookStackDataSource(client)
    coros = []
    for i in range(30):
        if i % 3 == 0:
            coros.append(datasource.update_content_permissions("page", i, owner_id=i))
        elif i % 3 == 1:
            coros.append(datasource.update_content_permissions("chapter", i, role_permissions=[{"role_id": i, "view": True}]))
        else:
            coros.append(datasource.update_content_permissions("book", i, fallback_permissions={"view": False}))
    results = await asyncio.gather(*coros)
    # Check that the correct field is present for each pattern
    for i, r in enumerate(results):
        if i % 3 == 0:
            pass
        elif i % 3 == 1:
            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.
#------------------------------------------------
import asyncio  # used to run async functions
# --- Function under test (EXACT COPY) ---
from typing import Dict, List, Optional, Union

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

# --- Minimal stubs/mocks for dependencies ---

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

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

# Minimal BookStackClient stub
class BookStackClient:
    def __init__(self, client):
        self.client = client

    def get_client(self):
        return self.client

# --- Mock HTTP layer for testing async calls ---
class MockHTTPClient:
    def __init__(self):
        self.headers = {"Authorization": "Token test:test", "Content-Type": "application/json"}
        self.executed_requests = []

    async def execute(self, request, **kwargs):
        # Simulate a successful update with echo of body and url
        self.executed_requests.append(request)
        # Return a mock response with the request data for verification
        return HTTPResponse({
            "url": request.url,
            "method": request.method,
            "headers": request.headers,
            "query_params": request.query_params,
            "body": request.body
        })

    def get_base_url(self):
        return "http://localhost"
from app.sources.external.bookstack.bookstack import BookStackDataSource

# --- Test Fixtures ---

@pytest.fixture
def mock_data_source():
    """Fixture to provide a BookStackDataSource with a mock HTTP client."""
    client = BookStackClient(MockHTTPClient())
    return BookStackDataSource(client)

# --- Basic Test Cases ---

@pytest.mark.asyncio
async def test_update_content_permissions_basic_success(mock_data_source):
    """Test basic successful permission update with all parameters."""
    result = await mock_data_source.update_content_permissions(
        content_type="page",
        content_id=42,
        owner_id=7,
        role_permissions=[{"role_id": 1, "view": True, "edit": False}],
        fallback_permissions={"view": True, "edit": False}
    )

@pytest.mark.asyncio
async def test_update_content_permissions_basic_minimal(mock_data_source):
    """Test minimal parameters (only required)."""
    result = await mock_data_source.update_content_permissions(
        content_type="book",
        content_id=123
    )

@pytest.mark.asyncio
async def test_update_content_permissions_basic_async_await(mock_data_source):
    """Test that function is awaitable and returns BookStackResponse."""
    codeflash_output = mock_data_source.update_content_permissions("chapter", 5); coro = codeflash_output
    result = await coro

# --- Edge Test Cases ---

@pytest.mark.asyncio
async def test_update_content_permissions_invalid_content_type(mock_data_source):
    """Test with an invalid content_type string."""
    result = await mock_data_source.update_content_permissions(
        content_type="not_a_valid_type",
        content_id=1
    )

@pytest.mark.asyncio
async def test_update_content_permissions_empty_role_permissions(mock_data_source):
    """Test with empty role_permissions list."""
    result = await mock_data_source.update_content_permissions(
        content_type="shelf",
        content_id=2,
        role_permissions=[]
    )

@pytest.mark.asyncio
async def test_update_content_permissions_none_fallback_permissions(mock_data_source):
    """Test with fallback_permissions as None (should not appear in body)."""
    result = await mock_data_source.update_content_permissions(
        content_type="book",
        content_id=2,
        fallback_permissions=None
    )

@pytest.mark.asyncio
async def test_update_content_permissions_concurrent_calls(mock_data_source):
    """Test concurrent execution for multiple different content items."""
    coros = [
        mock_data_source.update_content_permissions("page", 1, owner_id=10),
        mock_data_source.update_content_permissions("chapter", 2, owner_id=20),
        mock_data_source.update_content_permissions("book", 3, owner_id=30)
    ]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_update_content_permissions_exception_handling():
    """Test that exceptions in execute result in success=False and error set."""
    class FailingHTTPClient(MockHTTPClient):
        async def execute(self, request, **kwargs):
            raise RuntimeError("Simulated HTTP failure")

    client = BookStackClient(FailingHTTPClient())
    ds = BookStackDataSource(client)
    result = await ds.update_content_permissions("page", 1)

@pytest.mark.asyncio
async def test_update_content_permissions_large_id_and_long_strings(mock_data_source):
    """Test with very large content_id and long content_type string."""
    long_type = "x" * 200
    large_id = 10**12
    result = await mock_data_source.update_content_permissions(
        content_type=long_type,
        content_id=large_id,
        owner_id=999999999
    )

# --- Large Scale Test Cases ---

@pytest.mark.asyncio
async def test_update_content_permissions_many_concurrent_updates(mock_data_source):
    """Test many concurrent permission updates (scalability)."""
    N = 25  # Keep under 1000 for speed
    coros = [
        mock_data_source.update_content_permissions(
            content_type="page",
            content_id=i,
            owner_id=i*10,
            role_permissions=[{"role_id": i, "view": True}],
            fallback_permissions={"view": bool(i % 2)}
        )
        for i in range(N)
    ]
    results = await asyncio.gather(*coros)
    # Spot check a few
    for i in (0, N//2, N-1):
        pass

@pytest.mark.asyncio
async def test_update_content_permissions_large_role_permissions(mock_data_source):
    """Test with a large list of role_permissions."""
    role_permissions = [{"role_id": i, "view": True, "edit": False} for i in range(100)]
    result = await mock_data_source.update_content_permissions(
        content_type="chapter",
        content_id=123,
        role_permissions=role_permissions
    )

# --- Throughput Test Cases ---

@pytest.mark.asyncio
async def test_update_content_permissions_throughput_small_load(mock_data_source):
    """Throughput: small batch of updates."""
    coros = [
        mock_data_source.update_content_permissions("book", i)
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_update_content_permissions_throughput_medium_load(mock_data_source):
    """Throughput: medium batch of updates."""
    coros = [
        mock_data_source.update_content_permissions("page", i, owner_id=i)
        for i in range(20)
    ]
    results = await asyncio.gather(*coros)
    # Check that owner_id is echoed correctly
    for i, r in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_update_content_permissions_throughput_varied_parameters(mock_data_source):
    """Throughput: varied parameters in concurrent updates."""
    coros = []
    for i in range(10):
        if i % 2 == 0:
            coros.append(mock_data_source.update_content_permissions(
                "shelf", i, role_permissions=[{"role_id": i, "view": False}]))
        else:
            coros.append(mock_data_source.update_content_permissions(
                "book", i, fallback_permissions={"view": True, "edit": False}))
    results = await asyncio.gather(*coros)
    # Check correct body fields for each pattern
    for i, r in enumerate(results):
        if i % 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.update_content_permissions-mhbotbs8 and push.