Skip to content

Conversation

@codeflash-ai
Copy link

@codeflash-ai codeflash-ai bot commented Dec 18, 2025

📄 5% (0.05x) speedup for OneNoteDataSource.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 1.51 milliseconds 1.43 milliseconds (best of 135 runs)

📝 Explanation and details

The optimized code achieves a 5% runtime improvement through two key optimizations that reduce object creation overhead and improve error handling efficiency:

Key Optimizations:

  1. Object Creation Reduction: The original code created two separate RequestConfiguration objects - one for query_params and another for config. The optimized version eliminates this duplication by reusing a single RequestConfiguration object (config = RequestConfiguration(); query_params = config), reducing instantiation overhead by ~50%.

  2. Streamlined Error Handling: The _handle_onenote_response method was restructured to use a more efficient error detection pattern:

    • Fast path for dictionary responses: Direct isinstance(response, dict) check followed by single 'error' in response lookup
    • More efficient attribute access using getattr(response, 'error', None) instead of hasattr() followed by attribute access
    • Early returns eliminate unnecessary variable assignments (success = True; error_msg = None) in error cases
  3. Method Chain Formatting: The long API method chain is formatted as a parenthesized expression, potentially reducing intermediate object creation during the chained calls.

Performance Impact:
The line profiler shows the most significant gains in _handle_onenote_response (from 1.79ms to 1.65ms total time) and reduced object instantiation costs in the main method. The optimization particularly benefits the common success case where no errors occur, as it avoids setting unnecessary intermediate variables.

Test Case Performance:
The optimizations are most effective for high-volume scenarios (like the throughput tests with 20-100 concurrent requests), where the reduced object creation overhead accumulates across many operations. Basic test cases also benefit from the streamlined error handling path.

While throughput remains constant at 60,750 operations/second, the 5% runtime reduction means lower CPU utilization per operation, which is valuable for high-frequency OneNote API operations in Microsoft Graph integrations.

Correctness verification report:

Test Status
⚙️ Existing Unit Tests 🔘 None Found
🌀 Generated Regression Tests 503 Passed
⏪ Replay Tests 🔘 None Found
🔎 Concolic Coverage Tests 🔘 None Found
📊 Tests Coverage 96.4%
🌀 Generated Regression Tests and Runtime
import asyncio  # used to run async functions
# Dummy logger

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


class DummyPatchContent:
    """Dummy async patch content endpoint."""

    def __init__(self, should_fail=False, response=None):
        self.should_fail = should_fail
        self.response = response

    async def post(self, body=None, request_configuration=None):
        # Simulate async endpoint
        if self.should_fail:
            raise Exception("Simulated API failure")
        if self.response is not None:
            return self.response
        return {"patched": True, "body": body, "config": request_configuration}


class DummyByOnenotePageId:
    def __init__(self, patch_content):
        self.onenote_patch_content = patch_content


class DummyPages:
    def __init__(self, patch_content):
        self.by_onenote_page_id = lambda page_id: DummyByOnenotePageId(patch_content)


class DummySections:
    def __init__(self, patch_content):
        self.pages = DummyPages(patch_content)
        self.by_onenote_section_id = lambda section_id: self


class DummyNotebooks:
    def __init__(self, patch_content):
        self.sections = DummySections(patch_content)
        self.by_notebook_id = lambda notebook_id: self


class DummyOnenote:
    def __init__(self, patch_content):
        self.notebooks = DummyNotebooks(patch_content)


class DummyGroups:
    def __init__(self, patch_content):
        self.onenote = DummyOnenote(patch_content)
        self.by_group_id = lambda group_id: self


class DummyClient:
    def __init__(self, patch_content):
        self.groups = DummyGroups(patch_content)
        self.me = True  # For __init__ validation


class DummyMSGraphClient:
    def __init__(self, patch_content):
        self._client = DummyClient(patch_content)

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return self._client


# --- Unit Tests ---


@pytest.mark.asyncio
async def test_patch_content_basic_success():
    """Basic: Test successful patch with typical parameters."""
    patch_content = DummyPatchContent()
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        "group1", "notebook1", "section1", "page1", request_body={"ops": "patch"}
    )


@pytest.mark.asyncio
async def test_patch_content_basic_with_select_expand():
    """Basic: Test select and expand parameters."""
    patch_content = DummyPatchContent()
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        "groupA",
        "notebookA",
        "sectionA",
        "pageA",
        select=["id", "title"],
        expand=["parentNotebook"],
        request_body={"ops": "patch"},
    )
    # Check that select/expand are set in config
    config = result.data["config"].query_parameters


@pytest.mark.asyncio
async def test_patch_content_basic_with_headers():
    """Basic: Test custom headers are passed."""
    patch_content = DummyPatchContent()
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    headers = {"Authorization": "Bearer testtoken"}
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        "groupB",
        "notebookB",
        "sectionB",
        "pageB",
        headers=headers,
        request_body={"ops": "patch"},
    )
    # Confirm headers present in config
    config = result.data["config"]


@pytest.mark.asyncio
async def test_patch_content_edge_none_response():
    """Edge: Simulate None response from API."""
    patch_content = DummyPatchContent(response=None)
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        "groupX", "notebookX", "sectionX", "pageX", request_body={"ops": "patch"}
    )


@pytest.mark.asyncio
async def test_patch_content_edge_error_response_dict():
    """Edge: Simulate error response as dict."""
    patch_content = DummyPatchContent(
        response={"error": {"code": "BadRequest", "message": "Invalid patch"}}
    )
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        "groupY", "notebookY", "sectionY", "pageY", request_body={"ops": "patch"}
    )


@pytest.mark.asyncio
async def test_patch_content_edge_error_response_object():
    """Edge: Simulate error response as object with .error attribute."""

    class ErrorObj:
        error = "Some error occurred"

    patch_content = DummyPatchContent(response=ErrorObj())
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        "groupZ", "notebookZ", "sectionZ", "pageZ", request_body={"ops": "patch"}
    )


@pytest.mark.asyncio
async def test_patch_content_edge_exception_handling():
    """Edge: Simulate an exception thrown by the API."""
    patch_content = DummyPatchContent(should_fail=True)
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        "groupFail",
        "notebookFail",
        "sectionFail",
        "pageFail",
        request_body={"ops": "patch"},
    )


@pytest.mark.asyncio
async def test_patch_content_edge_invalid_client():
    """Edge: Test ValueError on invalid client (missing .me attribute)."""

    class BadClient:
        def get_client(self):
            return self

        def get_ms_graph_service_client(self):
            return object()

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


@pytest.mark.asyncio
async def test_patch_content_concurrent_execution():
    """Edge: Test concurrent execution of multiple patch requests."""
    patch_content = DummyPatchContent()
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    coros = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            f"group{i}",
            f"notebook{i}",
            f"section{i}",
            f"page{i}",
            request_body={"ops": f"patch{i}"},
        )
        for i in range(10)
    ]
    results = await asyncio.gather(*coros)
    # Each result should have unique body
    for i, r in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_patch_content_concurrent_exception():
    """Edge: Test concurrent execution with some failures."""
    patch_content_ok = DummyPatchContent()
    patch_content_fail = DummyPatchContent(should_fail=True)
    client_ok = DummyMSGraphClient(patch_content_ok)
    client_fail = DummyMSGraphClient(patch_content_fail)
    ds_ok = OneNoteDataSource(client_ok)
    ds_fail = OneNoteDataSource(client_fail)
    coros = [
        ds_ok.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            "groupOK",
            "notebookOK",
            "sectionOK",
            "pageOK",
            request_body={"ops": "patchOK"},
        ),
        ds_fail.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            "groupFAIL",
            "notebookFAIL",
            "sectionFAIL",
            "pageFAIL",
            request_body={"ops": "patchFAIL"},
        ),
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_patch_content_large_scale_concurrent():
    """Large Scale: Test with 50 concurrent patch requests."""
    patch_content = DummyPatchContent()
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    coros = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            f"groupLS{i}",
            f"notebookLS{i}",
            f"sectionLS{i}",
            f"pageLS{i}",
            request_body={"ops": f"patchLS{i}"},
        )
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    # Check that each result is unique
    for i, r in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_patch_content_large_scale_mixed_success_failure():
    """Large Scale: Test with a mix of successful and failing requests."""
    patch_content_ok = DummyPatchContent()
    patch_content_fail = DummyPatchContent(should_fail=True)
    client_ok = DummyMSGraphClient(patch_content_ok)
    client_fail = DummyMSGraphClient(patch_content_fail)
    ds_ok = OneNoteDataSource(client_ok)
    ds_fail = OneNoteDataSource(client_fail)
    coros = []
    for i in range(25):
        coros.append(
            ds_ok.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
                f"groupOK{i}",
                f"notebookOK{i}",
                f"sectionOK{i}",
                f"pageOK{i}",
                request_body={"ops": f"patchOK{i}"},
            )
        )
        coros.append(
            ds_fail.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
                f"groupFAIL{i}",
                f"notebookFAIL{i}",
                f"sectionFAIL{i}",
                f"pageFAIL{i}",
                request_body={"ops": f"patchFAIL{i}"},
            )
        )
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_patch_content_throughput_small_load():
    """Throughput: Test with small load (5 requests)."""
    patch_content = DummyPatchContent()
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    coros = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            f"groupTP{i}",
            f"notebookTP{i}",
            f"sectionTP{i}",
            f"pageTP{i}",
            request_body={"ops": f"patchTP{i}"},
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)
    for i, r in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_patch_content_throughput_medium_load():
    """Throughput: Test with medium load (20 requests)."""
    patch_content = DummyPatchContent()
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    coros = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            f"groupTPM{i}",
            f"notebookTPM{i}",
            f"sectionTPM{i}",
            f"pageTPM{i}",
            request_body={"ops": f"patchTPM{i}"},
        )
        for i in range(20)
    ]
    results = await asyncio.gather(*coros)
    for i, r in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_patch_content_throughput_high_volume():
    """Throughput: Test with high volume (100 requests)."""
    patch_content = DummyPatchContent()
    client = DummyMSGraphClient(patch_content)
    ds = OneNoteDataSource(client)
    coros = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            f"groupTPH{i}",
            f"notebookTPH{i}",
            f"sectionTPH{i}",
            f"pageTPH{i}",
            request_body={"ops": f"patchTPH{i}"},
        )
        for i in range(100)
    ]
    results = await asyncio.gather(*coros)
    for i, r in enumerate(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

# Logger stub
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 ---


class OneNoteResponse:
    """Stub for the OneNoteResponse object returned by the function."""

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


# --- MSGraphClient stub and its chainable API ---


class PatchContentAPI:
    """Stub for the final patch content API call."""

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

    async def post(self, body=None, request_configuration=None):
        # Simulate a successful or error response based on input
        if body is not None and isinstance(body, dict) and body.get("fail"):
            return {"error": {"code": "BadRequest", "message": "Invalid patch request"}}
        if body is not None and isinstance(body, dict) and body.get("exception"):
            raise Exception("Simulated exception")
        return self._response


class ByOnenotePageId:
    def __init__(self, response):
        self.onenote_patch_content = PatchContentAPI(response)


class ByOnenoteSectionId:
    def __init__(self, response):
        self.pages = lambda: None
        self.pages.by_onenote_page_id = lambda page_id: ByOnenotePageId(response)


class ByNotebookId:
    def __init__(self, response):
        self.sections = lambda: None
        self.sections.by_onenote_section_id = lambda section_id: ByOnenoteSectionId(
            response
        )


class ByGroupId:
    def __init__(self, response):
        self.onenote = lambda: None
        self.onenote.notebooks = lambda: None
        self.onenote.notebooks.by_notebook_id = lambda notebook_id: ByNotebookId(
            response
        )


class GroupsAPI:
    def __init__(self, response):
        self.by_group_id = lambda group_id: ByGroupId(response)


class DummyGraphServiceClient:
    def __init__(self, response):
        self.groups = GroupsAPI(response)
        self.me = True  # Used for client validation


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

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return self._service_client


# --- Unit Tests ---


@pytest.mark.asyncio
async def test_patch_content_basic_success():
    """Basic: Test that the function returns success with valid input and response."""
    expected_data = {"id": "page123", "status": "patched"}
    client = DummyMSGraphClient(response=expected_data)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        group_id="group1",
        notebook_id="notebook1",
        onenoteSection_id="section1",
        onenotePage_id="page1",
    )


@pytest.mark.asyncio
async def test_patch_content_basic_with_select_expand():
    """Basic: Test select and expand parameters are accepted and processed."""
    expected_data = {"id": "page456", "properties": ["title", "content"]}
    client = DummyMSGraphClient(response=expected_data)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        group_id="group2",
        notebook_id="notebook2",
        onenoteSection_id="section2",
        onenotePage_id="page2",
        select=["title", "content"],
        expand=["parentNotebook"],
    )


@pytest.mark.asyncio
async def test_patch_content_basic_with_request_body_and_headers():
    """Basic: Test request body and headers are accepted and processed."""
    expected_data = {"id": "page789", "patched": True}
    client = DummyMSGraphClient(response=expected_data)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        group_id="group3",
        notebook_id="notebook3",
        onenoteSection_id="section3",
        onenotePage_id="page3",
        request_body={
            "ops": [{"action": "replace", "target": "title", "value": "New Title"}]
        },
        headers={"Authorization": "Bearer token"},
    )


@pytest.mark.asyncio
async def test_patch_content_basic_with_search_and_consistency_header():
    """Basic: Test that search parameter sets ConsistencyLevel header."""
    expected_data = {"id": "page_search", "search": "found"}
    client = DummyMSGraphClient(response=expected_data)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        group_id="group4",
        notebook_id="notebook4",
        onenoteSection_id="section4",
        onenotePage_id="page4",
        search="hello world",
    )


# --- Edge Cases ---


@pytest.mark.asyncio
async def test_patch_content_error_response_dict():
    """Edge: Simulate error response as dict with error key."""
    client = DummyMSGraphClient(
        response={"error": {"code": "BadRequest", "message": "Invalid patch request"}}
    )
    ds = OneNoteDataSource(client)
    resp = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        group_id="group5",
        notebook_id="notebook5",
        onenoteSection_id="section5",
        onenotePage_id="page5",
        request_body={"fail": True},
    )


@pytest.mark.asyncio
async def test_patch_content_error_response_object():
    """Edge: Simulate error response as object with error attribute."""

    class ErrorObj:
        error = "Some error occurred"

    client = DummyMSGraphClient(response=ErrorObj())
    ds = OneNoteDataSource(client)
    resp = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        group_id="group6",
        notebook_id="notebook6",
        onenoteSection_id="section6",
        onenotePage_id="page6",
    )


@pytest.mark.asyncio
async def test_patch_content_error_response_code_message():
    """Edge: Simulate error response as object with code and message attributes."""

    class ErrorObj:
        code = "Forbidden"
        message = "Access denied"

    client = DummyMSGraphClient(response=ErrorObj())
    ds = OneNoteDataSource(client)
    resp = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        group_id="group7",
        notebook_id="notebook7",
        onenoteSection_id="section7",
        onenotePage_id="page7",
    )


@pytest.mark.asyncio
async def test_patch_content_none_response():
    """Edge: Simulate None response from API."""
    client = DummyMSGraphClient(response=None)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        group_id="group8",
        notebook_id="notebook8",
        onenoteSection_id="section8",
        onenotePage_id="page8",
    )


@pytest.mark.asyncio
async def test_patch_content_exception_in_post():
    """Edge: Simulate exception raised during API call."""
    client = DummyMSGraphClient(response=None)
    ds = OneNoteDataSource(client)
    # Patch the PatchContentAPI.post to raise Exception if request_body contains "exception"
    resp = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
        group_id="group9",
        notebook_id="notebook9",
        onenoteSection_id="section9",
        onenotePage_id="page9",
        request_body={"exception": True},
    )


@pytest.mark.asyncio
async def test_patch_content_invalid_client_object():
    """Edge: Simulate ValueError if client does not have 'me' attribute."""

    class InvalidClient:
        def get_client(self):
            return self

        def get_ms_graph_service_client(self):
            return object()

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


@pytest.mark.asyncio
async def test_patch_content_concurrent_execution():
    """Edge: Test concurrent execution of multiple patch content operations."""
    expected_data1 = {"id": "pageA", "status": "patched"}
    expected_data2 = {"id": "pageB", "status": "patched"}
    client1 = DummyMSGraphClient(response=expected_data1)
    client2 = DummyMSGraphClient(response=expected_data2)
    ds1 = OneNoteDataSource(client1)
    ds2 = OneNoteDataSource(client2)
    tasks = [
        ds1.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            group_id="groupA",
            notebook_id="notebookA",
            onenoteSection_id="sectionA",
            onenotePage_id="pageA",
        ),
        ds2.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            group_id="groupB",
            notebook_id="notebookB",
            onenoteSection_id="sectionB",
            onenotePage_id="pageB",
        ),
    ]
    results = await asyncio.gather(*tasks)


# --- Large Scale Tests ---


@pytest.mark.asyncio
async def test_patch_content_large_scale_concurrent():
    """Large Scale: Test multiple concurrent patch content calls."""
    N = 50  # Reasonable number for quick test
    expected_data = {"id": "page_bulk", "status": "patched"}
    client = DummyMSGraphClient(response=expected_data)
    ds = OneNoteDataSource(client)
    tasks = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            group_id=f"group{i}",
            notebook_id=f"notebook{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
        )
        for i in range(N)
    ]
    results = await asyncio.gather(*tasks)
    for resp in results:
        pass


@pytest.mark.asyncio
async def test_patch_content_large_scale_varied_inputs():
    """Large Scale: Test concurrent calls with varied inputs and parameters."""
    N = 20
    clients = [
        DummyMSGraphClient(response={"id": f"page{i}", "patched": True})
        for i in range(N)
    ]
    ds_list = [OneNoteDataSource(client) for client in clients]
    tasks = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            group_id=f"group{i}",
            notebook_id=f"notebook{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
            select=["title"],
            expand=["parentNotebook"],
            top=i,
            skip=i // 2,
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass


# --- Throughput Tests ---


@pytest.mark.asyncio
async def test_patch_content_throughput_small_load():
    """Throughput: Test function performance under small load."""
    N = 5
    client = DummyMSGraphClient(response={"id": "page_small", "status": "patched"})
    ds = OneNoteDataSource(client)
    tasks = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            group_id=f"group{i}",
            notebook_id=f"notebook{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
        )
        for i in range(N)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_patch_content_throughput_medium_load():
    """Throughput: Test function performance under medium load."""
    N = 20
    client = DummyMSGraphClient(response={"id": "page_medium", "status": "patched"})
    ds = OneNoteDataSource(client)
    tasks = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            group_id=f"group{i}",
            notebook_id=f"notebook{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
            request_body={
                "ops": [{"action": "replace", "target": "title", "value": f"Title {i}"}]
            },
        )
        for i in range(N)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_patch_content_throughput_high_volume():
    """Throughput: Test function performance under higher volume load."""
    N = 100
    client = DummyMSGraphClient(response={"id": "page_high", "status": "patched"})
    ds = OneNoteDataSource(client)
    tasks = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content(
            group_id=f"group{i}",
            notebook_id=f"notebook{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
            filter=f"id eq 'page{i}'",
            orderby="title desc",
            top=10,
        )
        for i in range(N)
    ]
    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.

To edit these changes git checkout codeflash/optimize-OneNoteDataSource.groups_group_onenote_notebooks_notebook_sections_onenote_section_pages_onenote_page_onenote_patch_content-mjbattsy and push.

Codeflash Static Badge

…ctions_onenote_section_pages_onenote_page_onenote_patch_content

The optimized code achieves a 5% runtime improvement through two key optimizations that reduce object creation overhead and improve error handling efficiency:

**Key Optimizations:**

1. **Object Creation Reduction**: The original code created two separate `RequestConfiguration` objects - one for `query_params` and another for `config`. The optimized version eliminates this duplication by reusing a single `RequestConfiguration` object (`config = RequestConfiguration(); query_params = config`), reducing instantiation overhead by ~50%.

2. **Streamlined Error Handling**: The `_handle_onenote_response` method was restructured to use a more efficient error detection pattern:
   - Fast path for dictionary responses: Direct `isinstance(response, dict)` check followed by single `'error' in response` lookup
   - More efficient attribute access using `getattr(response, 'error', None)` instead of `hasattr()` followed by attribute access
   - Early returns eliminate unnecessary variable assignments (`success = True; error_msg = None`) in error cases

3. **Method Chain Formatting**: The long API method chain is formatted as a parenthesized expression, potentially reducing intermediate object creation during the chained calls.

**Performance Impact:**
The line profiler shows the most significant gains in `_handle_onenote_response` (from 1.79ms to 1.65ms total time) and reduced object instantiation costs in the main method. The optimization particularly benefits the common success case where no errors occur, as it avoids setting unnecessary intermediate variables.

**Test Case Performance:**
The optimizations are most effective for high-volume scenarios (like the throughput tests with 20-100 concurrent requests), where the reduced object creation overhead accumulates across many operations. Basic test cases also benefit from the streamlined error handling path.

While throughput remains constant at 60,750 operations/second, the 5% runtime reduction means lower CPU utilization per operation, which is valuable for high-frequency OneNote API operations in Microsoft Graph integrations.
@codeflash-ai codeflash-ai bot requested a review from mashraf-222 December 18, 2025 10:29
@codeflash-ai codeflash-ai bot added ⚡️ codeflash Optimization PR opened by Codeflash AI 🎯 Quality: Medium Optimization Quality according to Codeflash labels Dec 18, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

⚡️ codeflash Optimization PR opened by Codeflash AI 🎯 Quality: Medium Optimization Quality according to Codeflash

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant