⚡️ Speed up method OneNoteDataSource.users_onenote_sections_update_pages_content by 12%

[codeflash-ai]

📄 12% (0.12x) speedup for OneNoteDataSource.users_onenote_sections_update_pages_content in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 499 microseconds → 447 microseconds (best of 5 runs)

📝 Explanation and details

The optimization achieves an 11% runtime improvement by eliminating unnecessary object instantiation when no query parameters are provided.

Key optimization: The original code unconditionally created RequestConfiguration() objects for both query_params and config, even when no query parameters were specified. The optimized version uses lazy initialization - it only creates these objects when actually needed.

Specific changes:

• Added has_query_params check using any() to determine if query parameter objects are needed
• Only instantiates RequestConfiguration() objects when parameters are present or when headers/search require configuration
• Reduced object creation overhead from ~360ns (194027+166184ns) to ~98ns when no query params are used

Performance impact: The line profiler shows the optimization is most effective for calls without query parameters - the common case where only user_id, section_id, and page_id are provided. In such scenarios, the code avoids creating two unnecessary RequestConfiguration objects, saving approximately 27% of the total execution time.

Throughput remains stable at 745 ops/sec, indicating the optimization doesn't affect concurrent processing capabilities while reducing per-operation overhead.

This optimization particularly benefits workloads with frequent simple OneNote page updates that don't require complex query parameters, which appears to be a common pattern based on the test cases focusing on basic parameter scenarios.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
# Logger stub

import pytest  # used for our unit tests
from app.sources.external.microsoft.one_note.one_note import OneNoteDataSource

# --- Minimal stub for MSGraphClient and its nested methods ---


class DummyContent:
    def __init__(self, put_result):
        self._put_result = put_result

    async def put(self, body=None, request_configuration=None):
        # Simulate async call returning whatever was configured
        if isinstance(self._put_result, Exception):
            raise self._put_result
        return self._put_result


class DummyPages:
    def __init__(self, put_result):
        self._put_result = put_result

    def by_onenote_page_id(self, onenotePage_id):
        return DummyContent(self._put_result)


class DummySections:
    def __init__(self, put_result):
        self._put_result = put_result

    def by_onenote_section_id(self, onenoteSection_id):
        return DummyPages(self._put_result)


class DummyOnenote:
    def __init__(self, put_result):
        self._put_result = put_result
        self.sections = DummySections(self._put_result)


class DummyUsers:
    def __init__(self, put_result):
        self._put_result = put_result

    def by_user_id(self, user_id):
        return DummyUser(self._put_result)


class DummyUser:
    def __init__(self, put_result):
        self._put_result = put_result
        self.onenote = DummyOnenote(self._put_result)


class DummyClient:
    """Simulates the MSGraphClient's get_client().get_ms_graph_service_client() chain."""

    def __init__(self, put_result):
        self.users = DummyUsers(put_result)
        self.me = True  # To pass hasattr(self.client, "me") check


class DummyMSGraphClient:
    def __init__(self, put_result):
        self._put_result = put_result

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return DummyClient(self._put_result)


# --- TESTS ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_basic_successful_update_returns_success():
    """Test that a successful update returns success=True and correct data."""

    # Simulate a successful response object
    class Response:
        pass

    response_obj = Response()
    client = DummyMSGraphClient(response_obj)
    ds = OneNoteDataSource(client)
    result = await ds.users_onenote_sections_update_pages_content(
        user_id="user1", onenoteSection_id="sectionA", onenotePage_id="pageX"
    )


@pytest.mark.asyncio
async def test_basic_error_response_with_error_attr():
    """Test that a response with an 'error' attribute is handled as error."""

    class Response:
        def __init__(self):
            self.error = "Some error!"

    response_obj = Response()
    client = DummyMSGraphClient(response_obj)
    ds = OneNoteDataSource(client)
    result = await ds.users_onenote_sections_update_pages_content(
        user_id="user2", onenoteSection_id="sectionB", onenotePage_id="pageY"
    )


@pytest.mark.asyncio
async def test_basic_error_response_with_error_dict():
    """Test that a dict response with an 'error' key is handled as error."""
    response_obj = {"error": {"code": "BadRequest", "message": "Invalid input"}}
    client = DummyMSGraphClient(response_obj)
    ds = OneNoteDataSource(client)
    result = await ds.users_onenote_sections_update_pages_content(
        user_id="user3", onenoteSection_id="sectionC", onenotePage_id="pageZ"
    )


@pytest.mark.asyncio
async def test_basic_error_response_with_code_and_message():
    """Test that a response with 'code' and 'message' attributes is handled as error."""

    class Response:
        def __init__(self):
            self.code = "Forbidden"
            self.message = "Access denied"

    response_obj = Response()
    client = DummyMSGraphClient(response_obj)
    ds = OneNoteDataSource(client)
    result = await ds.users_onenote_sections_update_pages_content(
        user_id="user4", onenoteSection_id="sectionD", onenotePage_id="pageW"
    )


@pytest.mark.asyncio
async def test_basic_empty_response_returns_error():
    """Test that a None response is handled as error."""
    response_obj = None
    client = DummyMSGraphClient(response_obj)
    ds = OneNoteDataSource(client)
    result = await ds.users_onenote_sections_update_pages_content(
        user_id="user5", onenoteSection_id="sectionE", onenotePage_id="pageV"
    )


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_concurrent_execution_returns_expected():
    """Test concurrent execution of the function with different inputs."""

    class ResponseA:
        pass

    class ResponseB:
        pass

    clientA = DummyMSGraphClient(ResponseA())
    clientB = DummyMSGraphClient(ResponseB())
    dsA = OneNoteDataSource(clientA)
    dsB = OneNoteDataSource(clientB)
    # Run two calls concurrently
    results = await asyncio.gather(
        dsA.users_onenote_sections_update_pages_content("u1", "s1", "p1"),
        dsB.users_onenote_sections_update_pages_content("u2", "s2", "p2"),
    )


@pytest.mark.asyncio
async def test_exception_in_put_is_handled():
    """Test that an exception in the async put call is handled and returns success=False."""
    client = DummyMSGraphClient(RuntimeError("Simulated failure"))
    ds = OneNoteDataSource(client)
    result = await ds.users_onenote_sections_update_pages_content(
        user_id="user6", onenoteSection_id="sectionF", onenotePage_id="pageU"
    )


@pytest.mark.asyncio
async def test_headers_and_search_sets_consistency_level():
    """Test that headers and search parameter set ConsistencyLevel header."""
    # We'll check that the config.headers has ConsistencyLevel set
    # by patching DummyContent.put to capture the config argument
    captured_config = {}

    class DummyContentCapture(DummyContent):
        async def put(self, body=None, request_configuration=None):
            captured_config["headers"] = getattr(request_configuration, "headers", None)
            return {"foo": "bar"}

    class DummyPagesCapture(DummyPages):
        def by_onenote_page_id(self, onenotePage_id):
            return DummyContentCapture(None)

    class DummySectionsCapture(DummySections):
        def by_onenote_section_id(self, onenoteSection_id):
            return DummyPagesCapture(None)

    class DummyOnenoteCapture(DummyOnenote):
        def __init__(self):
            self.sections = DummySectionsCapture(None)

    class DummyUserCapture(DummyUser):
        def __init__(self):
            self.onenote = DummyOnenoteCapture()

    class DummyUsersCapture(DummyUsers):
        def by_user_id(self, user_id):
            return DummyUserCapture(None)

    class DummyClientCapture(DummyClient):
        def __init__(self):
            self.users = DummyUsersCapture(None)
            self.me = True

    class DummyMSGraphClientCapture(DummyMSGraphClient):
        def get_client(self):
            return self

        def get_ms_graph_service_client(self):
            return DummyClientCapture()

    client = DummyMSGraphClientCapture(None)
    ds = OneNoteDataSource(client)
    await ds.users_onenote_sections_update_pages_content(
        user_id="user7",
        onenoteSection_id="sectionG",
        onenotePage_id="pageT",
        search="foo",
        headers={"X-Test": "yes"},
    )


@pytest.mark.asyncio
async def test_invalid_client_raises_valueerror():
    """Test that a client missing 'me' attribute raises ValueError on init."""

    class NoMeClient:
        def get_client(self):
            return self

        def get_ms_graph_service_client(self):
            return object()

    with pytest.raises(ValueError):
        OneNoteDataSource(NoMeClient())


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_large_scale_concurrent_updates():
    """Test multiple concurrent calls for scalability (10 parallel calls)."""

    class Response:
        def __init__(self, idx):
            self.idx = idx

    N = 10
    clients = [DummyMSGraphClient(Response(i)) for i in range(N)]
    ds_list = [OneNoteDataSource(c) for c in clients]
    coros = [
        ds.users_onenote_sections_update_pages_content(
            user_id=f"user{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*coros)


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_OneNoteDataSource_users_onenote_sections_update_pages_content_throughput_small_load():
    """Throughput: Test 5 quick concurrent updates (small load)."""

    class Response:
        def __init__(self, tag):
            self.tag = tag

    N = 5
    ds = OneNoteDataSource(DummyMSGraphClient(Response("ok")))
    coros = [
        ds.users_onenote_sections_update_pages_content(
            user_id=f"user{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
        )
        for i in range(N)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_OneNoteDataSource_users_onenote_sections_update_pages_content_throughput_medium_load():
    """Throughput: Test 25 concurrent updates (medium load)."""

    class Response:
        def __init__(self, tag):
            self.tag = tag

    N = 25
    ds = OneNoteDataSource(DummyMSGraphClient(Response("ok")))
    coros = [
        ds.users_onenote_sections_update_pages_content(
            user_id=f"user{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
        )
        for i in range(N)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_OneNoteDataSource_users_onenote_sections_update_pages_content_throughput_large_load():
    """Throughput: Test 100 concurrent updates (large but bounded load)."""

    class Response:
        def __init__(self, tag):
            self.tag = tag

    N = 100
    ds = OneNoteDataSource(DummyMSGraphClient(Response("ok")))
    coros = [
        ds.users_onenote_sections_update_pages_content(
            user_id=f"user{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
        )
        for i in range(N)
    ]
    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.

To edit these changes git checkout codeflash/optimize-OneNoteDataSource.users_onenote_sections_update_pages_content-mjf6jtoz and push.