⚡️ Speed up method OneNoteDataSource.groups_onenote_sections_pages_get_parent_notebook by 27%

[codeflash-ai]

📄 27% (0.27x) speedup for OneNoteDataSource.groups_onenote_sections_pages_get_parent_notebook in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 684 microseconds → 541 microseconds (best of 5 runs)

📝 Explanation and details

The optimization achieves a 26% runtime improvement by implementing lazy instantiation of Microsoft Graph SDK objects, avoiding unnecessary object creation when optional parameters are not provided.

Key optimizations applied:

1. Conditional object instantiation: Instead of always creating SectionsRequestBuilderGetQueryParameters() and SectionsRequestBuilderGetRequestConfiguration() objects, they are only created when needed. The code checks if any query parameters are actually provided before instantiation.

2. Early exit pattern: The optimization uses query_params = None and config = None as starting points, then conditionally creates these objects only when select, expand, filter, orderby, search, top, skip, or headers are provided.

Why this leads to speedup:

• Object creation overhead: The line profiler shows the original code spent 16.5% of time (390,751ns) just creating the query parameters object, and 8.8% (207,838ns) creating the configuration object on every call
• Reduced memory allocation: In the optimized version, when no parameters are provided (the common case), these expensive object creations are completely avoided
• CPU cache efficiency: Less object instantiation means better cache locality and reduced garbage collection pressure

Performance analysis from profiler data:

• Original: 2,369,540ns total time with heavy object creation costs
• Optimized: 1,553,450ns total time with conditional object creation
• The Microsoft Graph API call itself (48.9% vs 34.1% of total time) shows the optimization successfully reduces overhead around the actual network call

Test case performance:
The optimization is particularly effective for the common scenario where methods are called with minimal parameters. Most test cases show consistent improvements, especially the basic success cases and concurrent execution tests that represent typical usage patterns.

While throughput remains the same at 915 operations/second due to the async nature being I/O bound, the reduced CPU overhead per operation makes this optimization valuable for applications with high OneNote API call frequency.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio
from typing import Optional

import pytest
from app.sources.external.microsoft.one_note.one_note import OneNoteDataSource

# --- Minimal stubs for dependencies ---


class OneNoteResponse:
    """A simple OneNoteResponse stub for testing."""

    def __init__(self, success: bool, data=None, error: Optional[str] = None):
        self.success = success
        self.data = data
        self.error = error


# --- Minimal stub for MSGraphClient chain ---


class ParentNotebookStub:
    def __init__(self, response):
        self._response = response
        self._called = False

    async def get(self, request_configuration=None):
        self._called = True
        # Simulate an async call
        await asyncio.sleep(0)
        if isinstance(self._response, Exception):
            raise self._response
        return self._response


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

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


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

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


class OnenoteStub:
    def __init__(self, response):
        self._response = response
        self.sections = SectionsStub(response)


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

    def by_group_id(self, group_id):
        return GroupByIdStub(self._response)


class GroupByIdStub:
    def __init__(self, response):
        self._response = response
        self.onenote = OnenoteStub(response)


class MSGraphClientStub:
    def __init__(self, response):
        self.groups = GroupsStub(response)
        self.me = True  # for __init__ check

    def get_ms_graph_service_client(self):
        return self


# --- Stub for MSGraphClient wrapper ---


class MSGraphClient:
    def __init__(self, client):
        self.client = client

    def get_client(self):
        return self.client


# --- TESTS ---


# Helper to create a OneNoteDataSource with a given response
def make_datasource_with_response(response):
    msgraph_client = MSGraphClient(MSGraphClientStub(response))
    return OneNoteDataSource(msgraph_client)


@pytest.mark.asyncio
async def test_basic_success_response():
    """Test basic successful async call returns expected OneNoteResponse."""
    expected_data = {"id": "notebook123", "displayName": "My Notebook"}
    ds = make_datasource_with_response(expected_data)
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="g1", onenoteSection_id="s1", onenotePage_id="p1"
    )


@pytest.mark.asyncio
async def test_basic_error_response_dict():
    """Test error handling when response is a dict with 'error' key."""
    error_dict = {"error": {"code": "404", "message": "Not found"}}
    ds = make_datasource_with_response(error_dict)
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="g1", onenoteSection_id="s1", onenotePage_id="p1"
    )


@pytest.mark.asyncio
async def test_basic_error_response_attr():
    """Test error handling when response has an 'error' attribute."""

    class ErrorObj:
        error = "Something went wrong"

    ds = make_datasource_with_response(ErrorObj())
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="g1", onenoteSection_id="s1", onenotePage_id="p1"
    )


@pytest.mark.asyncio
async def test_basic_error_response_code_message():
    """Test error handling when response has 'code' and 'message' attributes."""

    class ErrorObj:
        code = "500"
        message = "Server error"

    ds = make_datasource_with_response(ErrorObj())
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="g1", onenoteSection_id="s1", onenotePage_id="p1"
    )


@pytest.mark.asyncio
async def test_basic_none_response():
    """Test when the response is None."""
    ds = make_datasource_with_response(None)
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="g1", onenoteSection_id="s1", onenotePage_id="p1"
    )


@pytest.mark.asyncio
async def test_select_and_expand_parameters():
    """Test that select and expand parameters are accepted and used."""
    expected_data = {"id": "notebook456"}
    ds = make_datasource_with_response(expected_data)
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="g2",
        onenoteSection_id="s2",
        onenotePage_id="p2",
        select=["id", "displayName"],
        expand=["sections"],
    )


@pytest.mark.asyncio
async def test_headers_and_search_set_consistency_level():
    """Test that headers and search trigger ConsistencyLevel header."""
    expected_data = {"id": "notebook789"}
    ds = make_datasource_with_response(expected_data)
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="g3",
        onenoteSection_id="s3",
        onenotePage_id="p3",
        headers={"Authorization": "Bearer token"},
        search="foo",
    )


@pytest.mark.asyncio
async def test_handles_exception_in_get():
    """Test that an exception in the async get call is handled."""
    ds = make_datasource_with_response(RuntimeError("Simulated failure"))
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="g4", onenoteSection_id="s4", onenotePage_id="p4"
    )


@pytest.mark.asyncio
async def test_handles_exception_in_handle_response():
    """Test that an exception in _handle_onenote_response is handled."""
    # Patch the instance to raise in _handle_onenote_response
    expected_data = object()
    ds = make_datasource_with_response(expected_data)

    def broken_handle_response(response):
        raise ValueError("broken handler")

    ds._handle_onenote_response = broken_handle_response
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="g5", onenoteSection_id="s5", onenotePage_id="p5"
    )


@pytest.mark.asyncio
async def test_concurrent_execution():
    """Test concurrent execution of multiple async calls."""
    expected_data1 = {"id": "nb1"}
    expected_data2 = {"id": "nb2"}
    ds1 = make_datasource_with_response(expected_data1)
    ds2 = make_datasource_with_response(expected_data2)
    # Run two calls concurrently
    results = await asyncio.gather(
        ds1.groups_onenote_sections_pages_get_parent_notebook(
            group_id="g1", onenoteSection_id="s1", onenotePage_id="p1"
        ),
        ds2.groups_onenote_sections_pages_get_parent_notebook(
            group_id="g2", onenoteSection_id="s2", onenotePage_id="p2"
        ),
    )


@pytest.mark.asyncio
async def test_edge_case_empty_strings():
    """Test handling of empty string parameters."""
    expected_data = {"id": "emptytest"}
    ds = make_datasource_with_response(expected_data)
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="", onenoteSection_id="", onenotePage_id=""
    )


@pytest.mark.asyncio
async def test_edge_case_large_select_expand():
    """Test with large select and expand lists (but <1000 elements)."""
    select = [f"prop{i}" for i in range(100)]
    expand = [f"expand{i}" for i in range(100)]
    expected_data = {"id": "large"}
    ds = make_datasource_with_response(expected_data)
    resp = await ds.groups_onenote_sections_pages_get_parent_notebook(
        group_id="g",
        onenoteSection_id="s",
        onenotePage_id="p",
        select=select,
        expand=expand,
    )


@pytest.mark.asyncio
async def test_large_scale_concurrent():
    """Test many concurrent async calls for scalability."""
    expected_data = {"id": "scale"}
    ds = make_datasource_with_response(expected_data)
    tasks = [
        ds.groups_onenote_sections_pages_get_parent_notebook(
            group_id=f"g{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
        )
        for i in range(20)
    ]
    results = await asyncio.gather(*tasks)
    for resp in results:
        pass


# --- THROUGHPUT TESTS ---


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_sections_pages_get_parent_notebook_throughput_small_load():
    """Throughput: Small load (5 concurrent calls)."""
    expected_data = {"id": "tp_small"}
    ds = make_datasource_with_response(expected_data)
    tasks = [
        ds.groups_onenote_sections_pages_get_parent_notebook(
            group_id=f"g{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_sections_pages_get_parent_notebook_throughput_medium_load():
    """Throughput: Medium load (25 concurrent calls)."""
    expected_data = {"id": "tp_medium"}
    ds = make_datasource_with_response(expected_data)
    tasks = [
        ds.groups_onenote_sections_pages_get_parent_notebook(
            group_id=f"g{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
        )
        for i in range(25)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_sections_pages_get_parent_notebook_throughput_large_load():
    """Throughput: Large load (100 concurrent calls)."""
    expected_data = {"id": "tp_large"}
    ds = make_datasource_with_response(expected_data)
    tasks = [
        ds.groups_onenote_sections_pages_get_parent_notebook(
            group_id=f"g{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
        )
        for i in range(100)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_sections_pages_get_parent_notebook_throughput_error_handling():
    """Throughput: Mix of success and error responses."""
    expected_data = {"id": "tp_mix"}
    ds_ok = make_datasource_with_response(expected_data)
    ds_err = make_datasource_with_response(RuntimeError("fail"))
    tasks = []
    for i in range(20):
        if i % 2 == 0:
            tasks.append(
                ds_ok.groups_onenote_sections_pages_get_parent_notebook(
                    group_id=f"g{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
                )
            )
        else:
            tasks.append(
                ds_err.groups_onenote_sections_pages_get_parent_notebook(
                    group_id=f"g{i}", onenoteSection_id=f"s{i}", onenotePage_id=f"p{i}"
                )
            )
    results = await asyncio.gather(*tasks)
    for i, resp 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.

To edit these changes git checkout codeflash/optimize-OneNoteDataSource.groups_onenote_sections_pages_get_parent_notebook-mjbioim6 and push.