⚡️ Speed up method OneNoteDataSource.me_onenote_section_groups_sections_update_pages by 10%

[codeflash-ai]

📄 10% (0.10x) speedup for OneNoteDataSource.me_onenote_section_groups_sections_update_pages in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 1.07 milliseconds → 970 microseconds (best of 22 runs)

📝 Explanation and details

The optimized code achieves a 9% runtime improvement and 4.8% throughput improvement through lazy initialization of query parameters.

Key optimization: Instead of always creating a RequestConfiguration() object for query parameters, the optimized version only creates it when at least one query parameter is actually provided (select, expand, filter, orderby, search, top, or skip).

Why this matters:

• The line profiler shows the original code always executes query_params = RequestConfiguration() (line with 513,891 nanoseconds, 13.1% of total time)
• In the optimized version, this expensive operation only runs when needed (just 5 times instead of 426 times in the profiler run)
• Most OneNote API calls don't use query parameters, making this a significant waste in the original implementation

Performance impact:

• Runtime: Reduced from 1.07ms to 970μs (9% faster)
• Throughput: Increased from 8,946 to 9,372 operations/second (4.8% improvement)
• The optimization is most effective for simple API calls without query parameters, which appear to be the majority based on test results

Test case benefits:

• Basic success/error response tests see the most improvement since they don't use query parameters
• Concurrent execution tests benefit significantly due to the multiplicative effect across many simultaneous calls
• Large-scale throughput tests (80-100 concurrent calls) show measurable improvements

This optimization is particularly valuable for high-frequency OneNote operations where the function may be called thousands of times with minimal query parameters, such as bulk page updates or status monitoring workflows.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # Used for running async functions and concurrent execution
from typing import Any, Optional

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


# --- Minimal stubs for dependencies to allow the function to run in tests ---
class OneNoteResponse:
    def __init__(self, success: bool, data: Any = None, error: Optional[str] = None):
        self.success = success
        self.data = data
        self.error = error


class DummyPatchResponse:
    def __init__(self, content=None, error=None, code=None, message=None):
        self.content = content
        self.error = error
        self.code = code
        self.message = message


class DummyPages:
    def __init__(self, page_id, patch_response):
        self.page_id = page_id
        self.patch_response = patch_response

    async def patch(self, body=None, request_configuration=None):
        # Simulate async patch operation
        return self.patch_response


class DummySections:
    def __init__(self, section_id, patch_response):
        self.section_id = section_id
        self.patch_response = patch_response

    def by_onenote_section_id(self, section_id):
        return DummyPages(section_id, self.patch_response)


class DummySectionGroups:
    def __init__(self, section_group_id, patch_response):
        self.section_group_id = section_group_id
        self.patch_response = patch_response

    def by_section_group_id(self, section_group_id):
        return DummySections(section_group_id, self.patch_response)


class DummyMe:
    def __init__(self, patch_response):
        self.onenote = DummyOneNote(patch_response)


class DummyOneNote:
    def __init__(self, patch_response):
        self.section_groups = DummySectionGroups("dummy_sg", patch_response)


class DummyMSGraphClient:
    def __init__(self, patch_response):
        self._patch_response = patch_response

    def get_client(self):
        return DummyMSGraphServiceClient(self._patch_response)


class DummyMSGraphServiceClient:
    def __init__(self, patch_response):
        self.me = DummyMe(patch_response)

    def get_ms_graph_service_client(self):
        return self


# --- Unit Tests ---


@pytest.mark.asyncio
async def test_basic_success_response():
    """Test basic async/await behavior and successful update."""
    patch_response = DummyPatchResponse(content={"id": "page123", "status": "updated"})
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg1",
        onenoteSection_id="sec1",
        onenotePage_id="pg1",
        request_body={"title": "New Title"},
    )


@pytest.mark.asyncio
async def test_basic_error_response_with_error_attr():
    """Test error handling when patch response has an 'error' attribute."""
    patch_response = DummyPatchResponse(error="Permission denied")
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg2", onenoteSection_id="sec2", onenotePage_id="pg2"
    )


@pytest.mark.asyncio
async def test_basic_error_response_with_code_and_message():
    """Test error handling when patch response has 'code' and 'message' attributes."""
    patch_response = DummyPatchResponse(code="404", message="Not Found")
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg3", onenoteSection_id="sec3", onenotePage_id="pg3"
    )


@pytest.mark.asyncio
async def test_basic_error_response_with_dict_error():
    """Test error handling when patch response is a dict with 'error' key."""
    patch_response = {"error": {"code": "500", "message": "Internal Server Error"}}
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg4", onenoteSection_id="sec4", onenotePage_id="pg4"
    )


@pytest.mark.asyncio
async def test_basic_none_response():
    """Test error handling when patch response is None."""
    patch_response = None
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg5", onenoteSection_id="sec5", onenotePage_id="pg5"
    )


@pytest.mark.asyncio
async def test_select_and_expand_parameters():
    """Test select and expand parameters are handled correctly."""
    patch_response = DummyPatchResponse(content={"id": "pageX", "status": "updated"})
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg6",
        onenoteSection_id="sec6",
        onenotePage_id="pg6",
        select=["id", "title"],
        expand=["parentNotebook"],
        request_body={"title": "Updated"},
    )


@pytest.mark.asyncio
async def test_headers_and_search_consistency_level():
    """Test that headers and search add ConsistencyLevel header."""
    patch_response = DummyPatchResponse(content={"id": "pageY", "status": "updated"})
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    # Provide headers and search
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg7",
        onenoteSection_id="sec7",
        onenotePage_id="pg7",
        headers={"Authorization": "Bearer xyz"},
        search="meeting notes",
    )


@pytest.mark.asyncio
async def test_filter_orderby_top_skip_parameters():
    """Test filter, orderby, top, and skip parameters are handled."""
    patch_response = DummyPatchResponse(content={"id": "pageZ", "status": "updated"})
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg8",
        onenoteSection_id="sec8",
        onenotePage_id="pg8",
        filter="status eq 'active'",
        orderby="createdDateTime desc",
        top=5,
        skip=2,
    )


@pytest.mark.asyncio
async def test_concurrent_execution():
    """Test concurrent execution with asyncio.gather()."""
    patch_response1 = DummyPatchResponse(content={"id": "pageA", "status": "updated"})
    patch_response2 = DummyPatchResponse(content={"id": "pageB", "status": "updated"})
    client1 = DummyMSGraphClient(patch_response1)
    client2 = DummyMSGraphClient(patch_response2)
    ds1 = OneNoteDataSource(client1)
    ds2 = OneNoteDataSource(client2)
    results = await asyncio.gather(
        ds1.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id="sgA", onenoteSection_id="secA", onenotePage_id="pgA"
        ),
        ds2.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id="sgB", onenoteSection_id="secB", onenotePage_id="pgB"
        ),
    )


@pytest.mark.asyncio
async def test_large_scale_concurrent_calls():
    """Test function scalability with multiple concurrent calls."""
    patch_response = DummyPatchResponse(content={"id": "pageBulk", "status": "updated"})
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    tasks = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sg{i}",
            onenoteSection_id=f"sec{i}",
            onenotePage_id=f"pg{i}",
        )
        for i in range(20)  # 20 concurrent calls
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_large_scale_concurrent_error_calls():
    """Test function scalability with multiple concurrent error calls."""
    patch_response = DummyPatchResponse(error="Bulk error")
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    tasks = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sgE{i}",
            onenoteSection_id=f"secE{i}",
            onenotePage_id=f"pgE{i}",
        )
        for i in range(15)  # 15 concurrent error calls
    ]
    results = await asyncio.gather(*tasks)


# --- Throughput Tests ---


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_sections_update_pages_throughput_small_load():
    """Throughput test: small load (5 concurrent calls)."""
    patch_response = DummyPatchResponse(
        content={"id": "pageSmall", "status": "updated"}
    )
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    tasks = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sgS{i}",
            onenoteSection_id=f"secS{i}",
            onenotePage_id=f"pgS{i}",
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_sections_update_pages_throughput_medium_load():
    """Throughput test: medium load (30 concurrent calls)."""
    patch_response = DummyPatchResponse(
        content={"id": "pageMedium", "status": "updated"}
    )
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    tasks = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sgM{i}",
            onenoteSection_id=f"secM{i}",
            onenotePage_id=f"pgM{i}",
        )
        for i in range(30)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_sections_update_pages_throughput_high_volume():
    """Throughput test: high volume (80 concurrent calls)."""
    patch_response = DummyPatchResponse(content={"id": "pageHigh", "status": "updated"})
    client = DummyMSGraphClient(patch_response)
    ds = OneNoteDataSource(client)
    tasks = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sgH{i}",
            onenoteSection_id=f"secH{i}",
            onenotePage_id=f"pgH{i}",
        )
        for i in range(80)
    ]
    results = await asyncio.gather(*tasks)


# 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 pytest  # used for our unit tests
from app.sources.external.microsoft.one_note.one_note import OneNoteDataSource


class DummyPatch:
    def __init__(self, response):
        self._response = response

    async def patch(self, body=None, request_configuration=None):
        # Simulate async patch call
        return self._response


class DummyPages:
    def __init__(self, response):
        self._response = response

    def by_onenote_page_id(self, onenotePage_id):
        return DummyPatch(self._response)


class DummySections:
    def __init__(self, response):
        self._response = response

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


class DummySectionGroups:
    def __init__(self, response):
        self._response = response

    def by_section_group_id(self, sectionGroup_id):
        return DummySections(self._response)


class DummyMe:
    def __init__(self, response):
        self._response = response
        self.onenote = DummyOneNote(self._response)


class DummyOneNote:
    def __init__(self, response):
        self._response = response
        self.section_groups = DummySectionGroups(self._response)


class DummyMSGraphServiceClient:
    def __init__(self, response):
        self.me = DummyMe(response)


class DummyMSGraphClient:
    def __init__(self, response):
        self._response = response

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return DummyMSGraphServiceClient(self._response)


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

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_update_pages_basic_success():
    """Test basic successful update with minimal required parameters."""

    # Simulate a successful response object
    class ResponseObj:
        pass

    response = ResponseObj()
    ds = OneNoteDataSource(DummyMSGraphClient(response))
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg1", onenoteSection_id="s1", onenotePage_id="p1"
    )


@pytest.mark.asyncio
async def test_update_pages_basic_with_all_parameters():
    """Test update with all parameters provided."""

    class ResponseObj:
        pass

    response = ResponseObj()
    ds = OneNoteDataSource(DummyMSGraphClient(response))
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg2",
        onenoteSection_id="s2",
        onenotePage_id="p2",
        select=["id", "title"],
        expand=["parentNotebook"],
        filter="title eq 'Test'",
        orderby="title",
        search="notes",
        top=10,
        skip=2,
        request_body={"title": "Updated"},
        headers={"Authorization": "Bearer token"},
    )


@pytest.mark.asyncio
async def test_update_pages_basic_select_as_string():
    """Test select parameter as a single string instead of list."""

    class ResponseObj:
        pass

    response = ResponseObj()
    ds = OneNoteDataSource(DummyMSGraphClient(response))
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg3", onenoteSection_id="s3", onenotePage_id="p3", select="id"
    )


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_update_pages_error_response_object():
    """Test when response object has an error attribute."""

    class ResponseObj:
        error = "Some error occurred"

    response = ResponseObj()
    ds = OneNoteDataSource(DummyMSGraphClient(response))
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg4", onenoteSection_id="s4", onenotePage_id="p4"
    )


@pytest.mark.asyncio
async def test_update_pages_error_response_dict():
    """Test when response is a dict with 'error' key."""
    response = {"error": {"code": "BadRequest", "message": "Invalid data"}}
    ds = OneNoteDataSource(DummyMSGraphClient(response))
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg5", onenoteSection_id="s5", onenotePage_id="p5"
    )


@pytest.mark.asyncio
async def test_update_pages_error_response_dict_str():
    """Test when response is a dict with 'error' as a string."""
    response = {"error": "Some error string"}
    ds = OneNoteDataSource(DummyMSGraphClient(response))
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg6", onenoteSection_id="s6", onenotePage_id="p6"
    )


@pytest.mark.asyncio
async def test_update_pages_error_response_code_message():
    """Test when response has code and message attributes."""

    class ResponseObj:
        code = "404"
        message = "Not found"

    response = ResponseObj()
    ds = OneNoteDataSource(DummyMSGraphClient(response))
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg7", onenoteSection_id="s7", onenotePage_id="p7"
    )


@pytest.mark.asyncio
async def test_update_pages_none_response():
    """Test when response is None."""
    response = None
    ds = OneNoteDataSource(DummyMSGraphClient(response))
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg8", onenoteSection_id="s8", onenotePage_id="p8"
    )


@pytest.mark.asyncio
async def test_update_pages_exception_in_patch():
    """Test exception thrown during patch call."""

    class FailingPatch:
        async def patch(self, body=None, request_configuration=None):
            raise RuntimeError("Patch failed")

    class FailingPages:
        def by_onenote_page_id(self, onenotePage_id):
            return FailingPatch()

    class FailingSections:
        def by_onenote_section_id(self, onenoteSection_id):
            return FailingPages()

    class FailingSectionGroups:
        def by_section_group_id(self, sectionGroup_id):
            return FailingSections()

    class FailingMe:
        def __init__(self):
            self.onenote = FailingOneNote()

    class FailingOneNote:
        def __init__(self):
            self.section_groups = FailingSectionGroups()

    class FailingClient:
        def __init__(self):
            self.me = FailingMe()

    class FailingMSGraphServiceClient:
        def __init__(self):
            self.me = FailingMe()

    class FailingMSGraphClient:
        def get_client(self):
            return self

        def get_ms_graph_service_client(self):
            return FailingMSGraphServiceClient()

    ds = OneNoteDataSource(FailingMSGraphClient())
    result = await ds.me_onenote_section_groups_sections_update_pages(
        sectionGroup_id="sg9", onenoteSection_id="s9", onenotePage_id="p9"
    )


@pytest.mark.asyncio
async def test_update_pages_concurrent_execution():
    """Test multiple concurrent executions with different responses."""

    class ResponseObj1:
        pass

    class ResponseObj2:
        error = "Error in second"

    response1 = ResponseObj1()
    response2 = ResponseObj2()
    ds1 = OneNoteDataSource(DummyMSGraphClient(response1))
    ds2 = OneNoteDataSource(DummyMSGraphClient(response2))
    results = await asyncio.gather(
        ds1.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id="sg10", onenoteSection_id="s10", onenotePage_id="p10"
        ),
        ds2.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id="sg11", onenoteSection_id="s11", onenotePage_id="p11"
        ),
    )


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_update_pages_large_scale_concurrent():
    """Test large scale concurrent execution for robustness."""

    class ResponseObj:
        pass

    # Create 50 concurrent updates
    ds_list = [OneNoteDataSource(DummyMSGraphClient(ResponseObj())) for _ in range(50)]
    coros = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sg{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_update_pages_large_scale_mixed_errors():
    """Test large scale concurrent execution with mixed error/success responses."""

    class SuccessResponse:
        pass

    class ErrorResponse:
        error = "Bulk error"

    ds_list = []
    for i in range(25):
        ds_list.append(OneNoteDataSource(DummyMSGraphClient(SuccessResponse())))
    for i in range(25):
        ds_list.append(OneNoteDataSource(DummyMSGraphClient(ErrorResponse())))
    coros = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sg{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        if i < 25:
            pass
        else:
            pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_update_pages_throughput_small_load():
    """Throughput test: small load, 5 concurrent requests."""

    class ResponseObj:
        pass

    ds_list = [OneNoteDataSource(DummyMSGraphClient(ResponseObj())) for _ in range(5)]
    coros = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sg{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_update_pages_throughput_medium_load():
    """Throughput test: medium load, 20 concurrent requests."""

    class ResponseObj:
        pass

    ds_list = [OneNoteDataSource(DummyMSGraphClient(ResponseObj())) for _ in range(20)]
    coros = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sg{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_update_pages_throughput_high_volume():
    """Throughput test: high volume, 100 concurrent requests."""

    class ResponseObj:
        pass

    ds_list = [OneNoteDataSource(DummyMSGraphClient(ResponseObj())) for _ in range(100)]
    coros = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sg{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_update_pages_throughput_mixed_load():
    """Throughput test: mixed load, 30 requests (half error, half success)."""

    class SuccessResponse:
        pass

    class ErrorResponse:
        error = "Load error"

    ds_list = []
    for i in range(15):
        ds_list.append(OneNoteDataSource(DummyMSGraphClient(SuccessResponse())))
    for i in range(15):
        ds_list.append(OneNoteDataSource(DummyMSGraphClient(ErrorResponse())))
    coros = [
        ds.me_onenote_section_groups_sections_update_pages(
            sectionGroup_id=f"sg{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        if i < 15:
            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.

To edit these changes git checkout codeflash/optimize-OneNoteDataSource.me_onenote_section_groups_sections_update_pages-mjg4z170 and push.