⚡️ Speed up method OneNoteDataSource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section by 8%

[codeflash-ai]

📄 8% (0.08x) speedup for OneNoteDataSource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 1.98 milliseconds → 1.84 milliseconds (best of 82 runs)

📝 Explanation and details

The optimized code achieves a 7% runtime improvement through strategic object creation optimization and control flow simplification, while maintaining identical throughput and functionality.

Key optimizations applied:

1. Conditional object creation in async method: Instead of always creating a RequestConfiguration() object for query parameters, the optimized version uses a dictionary (qp) to collect parameters first, then only creates the RequestConfiguration object if parameters are actually present. This eliminates unnecessary object instantiation in the common case where no query parameters are provided.

2. Early returns in error handling: The _handle_onenote_response method now uses early returns instead of setting intermediate variables (success, error_msg) and branching logic. This reduces the number of variable assignments and conditional checks in the hot path.

3. Defensive header copying: The optimized version creates a copy of headers when provided (headers.copy()) to prevent potential mutation of the caller's input, which is a safer programming practice.

4. Method chain extraction: The long method chain is now assigned to an intermediate variable before calling post(), improving code readability without affecting performance.

Performance analysis from line profiler:

• The async method shows reduced time in the query parameter setup phase (lines building RequestConfiguration)
• The _handle_onenote_response method benefits from fewer variable assignments and more direct return paths
• Overall function call overhead is reduced through fewer object instantiations

Test case performance: The optimization performs well across all test scenarios - basic operations, error handling, concurrent execution, and high-volume throughput tests all maintain correctness while benefiting from the reduced overhead.

These optimizations are particularly beneficial for high-frequency OneNote API operations where the parameter setup overhead can accumulate, making the 7% improvement valuable for applications with intensive OneNote integration workflows.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio

# Dummy logger
from typing import Any, Optional

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

# --- Minimal stubs for dependencies and response types ---


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


# --- Dummy MSGraphClient and related chainable objects ---


class DummyCopyToSection:
    def __init__(self, response):
        self._response = response
        self.called_with = None

    async def post(self, body=None, request_configuration=None):
        self.called_with = (body, request_configuration)
        # Simulate error if body is a special value
        if body == "raise":
            raise Exception("Simulated API error")
        return self._response


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

    def by_onenote_page_id(self, onenotePage_id):
        return self

    @property
    def copy_to_section(self):
        return DummyCopyToSection(self._response)


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

    def by_onenote_section_id(self, onenoteSection_id):
        return self

    @property
    def pages(self):
        return DummyPages(self._response)


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

    def by_section_group_id(self, sectionGroup_id):
        return self

    @property
    def sections(self):
        return DummySections(self._response)


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

    def by_notebook_id(self, notebook_id):
        return self

    @property
    def section_groups(self):
        return DummySectionGroups(self._response)


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

    @property
    def onenote(self):
        return self

    @property
    def notebooks(self):
        return DummyNotebooks(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)


# --- TESTS ---

# 1. BASIC TEST CASES


@pytest.mark.asyncio
async def test_copy_to_section_basic_success():
    """Test basic successful call with required parameters only."""
    dummy_response = {"id": "page123", "status": "copied"}
    client = DummyMSGraphClient(dummy_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="nb1",
        sectionGroup_id="sg1",
        onenoteSection_id="sec1",
        onenotePage_id="page1",
    )


@pytest.mark.asyncio
async def test_copy_to_section_basic_with_all_parameters():
    """Test with all optional parameters provided."""
    dummy_response = {"id": "page456", "status": "copied"}
    client = DummyMSGraphClient(dummy_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="nb2",
        sectionGroup_id="sg2",
        onenoteSection_id="sec2",
        onenotePage_id="page2",
        select=["id", "status"],
        expand=["parentSection"],
        filter="status eq 'active'",
        orderby="createdDateTime desc",
        search="important",
        top=10,
        skip=2,
        request_body={"destinationId": "destSection"},
        headers={"Authorization": "Bearer token"},
    )


@pytest.mark.asyncio
async def test_copy_to_section_empty_response():
    """Test that a None response is handled as an error."""
    client = DummyMSGraphClient(None)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="nb3",
        sectionGroup_id="sg3",
        onenoteSection_id="sec3",
        onenotePage_id="page3",
    )


@pytest.mark.asyncio
async def test_copy_to_section_error_in_response_dict():
    """Test that a response with an 'error' dict is handled as error."""
    error_response = {"error": {"code": "400", "message": "Invalid request"}}
    client = DummyMSGraphClient(error_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="nb4",
        sectionGroup_id="sg4",
        onenoteSection_id="sec4",
        onenotePage_id="page4",
    )


# 2. EDGE TEST CASES


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

    class ErrorObj:
        error = "Something went wrong"

    client = DummyMSGraphClient(ErrorObj())
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="nb5",
        sectionGroup_id="sg5",
        onenoteSection_id="sec5",
        onenotePage_id="page5",
    )


@pytest.mark.asyncio
async def test_copy_to_section_exception_handling():
    """Test that exceptions in the API call are caught and handled."""
    # Use a special request_body value to trigger an exception in DummyCopyToSection.post
    client = DummyMSGraphClient({"id": "should not be returned"})
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="nb6",
        sectionGroup_id="sg6",
        onenoteSection_id="sec6",
        onenotePage_id="page6",
        request_body="raise",
    )


@pytest.mark.asyncio
async def test_copy_to_section_concurrent_execution():
    """Test concurrent execution of the async function."""
    client = DummyMSGraphClient({"id": "page_concurrent", "status": "copied"})
    ds = OneNoteDataSource(client)

    async def call_it(i):
        return await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id=f"nb{i}",
            sectionGroup_id=f"sg{i}",
            onenoteSection_id=f"sec{i}",
            onenotePage_id=f"page{i}",
        )

    tasks = [call_it(i) for i in range(5)]
    results = await asyncio.gather(*tasks)
    for res in results:
        pass


@pytest.mark.asyncio
async def test_copy_to_section_search_adds_consistency_header():
    """Test that search parameter adds ConsistencyLevel header."""
    dummy_response = {"id": "page_search", "status": "copied"}
    client = DummyMSGraphClient(dummy_response)
    ds = OneNoteDataSource(client)
    # Patch the config object to check header
    captured_config = {}
    orig_post = DummyCopyToSection.post

    async def patched_post(self, body=None, request_configuration=None):
        captured_config["headers"] = getattr(request_configuration, "headers", None)
        return await orig_post(self, body, request_configuration)

    DummyCopyToSection.post = patched_post

    try:
        await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id="nb7",
            sectionGroup_id="sg7",
            onenoteSection_id="sec7",
            onenotePage_id="page7",
            search="findme",
        )
    finally:
        DummyCopyToSection.post = orig_post  # Restore


# 3. LARGE SCALE TEST CASES


@pytest.mark.asyncio
async def test_copy_to_section_large_scale_concurrent():
    """Test function under moderate concurrent load (50 calls)."""
    dummy_response = {"id": "page_large", "status": "copied"}
    client = DummyMSGraphClient(dummy_response)
    ds = OneNoteDataSource(client)

    async def call_it(i):
        return await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id=f"nb{i}",
            sectionGroup_id=f"sg{i}",
            onenoteSection_id=f"sec{i}",
            onenotePage_id=f"page{i}",
        )

    tasks = [call_it(i) for i in range(50)]
    results = await asyncio.gather(*tasks)


# 4. THROUGHPUT TEST CASES


@pytest.mark.asyncio
async def test_copy_to_section_throughput_small_load():
    """Throughput: test 5 concurrent requests complete quickly and correctly."""
    dummy_response = {"id": "page_throughput_small", "status": "copied"}
    client = DummyMSGraphClient(dummy_response)
    ds = OneNoteDataSource(client)

    async def call_it(i):
        return await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id=f"nb{i}",
            sectionGroup_id=f"sg{i}",
            onenoteSection_id=f"sec{i}",
            onenotePage_id=f"page{i}",
        )

    tasks = [call_it(i) for i in range(5)]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_copy_to_section_throughput_medium_load():
    """Throughput: test 20 concurrent requests."""
    dummy_response = {"id": "page_throughput_medium", "status": "copied"}
    client = DummyMSGraphClient(dummy_response)
    ds = OneNoteDataSource(client)

    async def call_it(i):
        return await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id=f"nb{i}",
            sectionGroup_id=f"sg{i}",
            onenoteSection_id=f"sec{i}",
            onenotePage_id=f"page{i}",
        )

    tasks = [call_it(i) for i in range(20)]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_copy_to_section_throughput_large_load():
    """Throughput: test 100 concurrent requests."""
    dummy_response = {"id": "page_throughput_large", "status": "copied"}
    client = DummyMSGraphClient(dummy_response)
    ds = OneNoteDataSource(client)

    async def call_it(i):
        return await ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id=f"nb{i}",
            sectionGroup_id=f"sg{i}",
            onenoteSection_id=f"sec{i}",
            onenotePage_id=f"page{i}",
        )

    tasks = [call_it(i) for i in range(100)]
    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
from typing import Any, Optional

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


# Minimal stub classes for testing
class OneNoteResponse:
    def __init__(self, success: bool, data: Any = None, error: Optional[str] = None):
        self.success = success
        self.data = data
        self.error = error


class DummyCopyToSection:
    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 POST call
        if self.should_fail:
            raise Exception("Simulated failure in post")
        if self.response is not None:
            return self.response
        # Return a dummy successful response
        return {"copied": True, "body": body, "config": request_configuration}


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

    def by_onenote_page_id(self, onenotePage_id):
        return DummyCopyToSection(self.should_fail, self.response)


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

    def by_onenote_section_id(self, onenoteSection_id):
        return DummyPages(self.should_fail, self.response)


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

    def by_section_group_id(self, sectionGroup_id):
        return DummySections(self.should_fail, self.response)


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

    def by_notebook_id(self, notebook_id):
        return DummySectionGroups(self.should_fail, self.response)


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

    @property
    def onenote(self):
        return self

    @property
    def notebooks(self):
        return DummyNotebooks(self.should_fail, self.response)


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

    @property
    def me(self):
        return DummyMe(self.should_fail, self.response)

    def get_ms_graph_service_client(self):
        return self


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

    def get_client(self):
        return DummyClient(self.should_fail, self.response)


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

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_copy_to_section_basic_success():
    """Test basic successful copy operation with required parameters."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    result = await datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="notebook123",
        sectionGroup_id="sg456",
        onenoteSection_id="section789",
        onenotePage_id="page001",
    )


@pytest.mark.asyncio
async def test_copy_to_section_with_all_parameters():
    """Test operation with all optional parameters set."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    result = await datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="notebookA",
        sectionGroup_id="sgB",
        onenoteSection_id="sectionC",
        onenotePage_id="pageD",
        select=["id", "title"],
        expand=["parentSection"],
        filter="title eq 'Test'",
        orderby="createdDateTime desc",
        search="meeting notes",
        top=10,
        skip=2,
        request_body={"destinationId": "sectionC"},
        headers={"Custom-Header": "Value"},
    )


@pytest.mark.asyncio
async def test_copy_to_section_empty_response():
    """Test handling of None response from API."""
    # Simulate None response from post
    datasource = OneNoteDataSource(DummyMSGraphClient(response=None))
    result = await datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="notebookX",
        sectionGroup_id="sgY",
        onenoteSection_id="sectionZ",
        onenotePage_id="pageW",
    )


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_copy_to_section_api_error_dict():
    """Test handling of error in dict response."""
    error_response = {"error": {"code": "BadRequest", "message": "Invalid section"}}
    datasource = OneNoteDataSource(DummyMSGraphClient(response=error_response))
    result = await datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="notebookErr",
        sectionGroup_id="sgErr",
        onenoteSection_id="sectionErr",
        onenotePage_id="pageErr",
    )


@pytest.mark.asyncio
async def test_copy_to_section_api_error_object():
    """Test handling of error in object response."""

    class ErrorObj:
        error = "API failed"

    datasource = OneNoteDataSource(DummyMSGraphClient(response=ErrorObj()))
    result = await datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="notebookObj",
        sectionGroup_id="sgObj",
        onenoteSection_id="sectionObj",
        onenotePage_id="pageObj",
    )


@pytest.mark.asyncio
async def test_copy_to_section_api_exception():
    """Test handling of exception thrown by post method."""
    datasource = OneNoteDataSource(DummyMSGraphClient(should_fail=True))
    result = await datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
        notebook_id="notebookEx",
        sectionGroup_id="sgEx",
        onenoteSection_id="sectionEx",
        onenotePage_id="pageEx",
    )


@pytest.mark.asyncio
async def test_copy_to_section_concurrent_execution():
    """Test concurrent execution of multiple copy operations."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    tasks = [
        datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id=f"notebook{i}",
            sectionGroup_id=f"sg{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
        )
        for i in range(10)
    ]
    results = await asyncio.gather(*tasks)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_copy_to_section_invalid_client():
    """Test initialization with an invalid client (missing 'me' attribute)."""

    class BadClient:
        def get_ms_graph_service_client(self):
            return self

    class BadMSGraphClient:
        def get_client(self):
            return BadClient()

    with pytest.raises(ValueError) as excinfo:
        OneNoteDataSource(BadMSGraphClient())


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_copy_to_section_large_scale_concurrent():
    """Test large scale concurrent copy operations (up to 100)."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    tasks = [
        datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id=f"notebook{i}",
            sectionGroup_id=f"sg{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
        )
        for i in range(100)
    ]
    results = await asyncio.gather(*tasks)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_copy_to_section_large_scale_with_errors():
    """Test large scale concurrent copy operations with some errors."""
    # Every 10th operation will fail
    datasources = [
        OneNoteDataSource(DummyMSGraphClient(should_fail=(i % 10 == 0)))
        for i in range(50)
    ]
    tasks = [
        ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id=f"notebook{i}",
            sectionGroup_id=f"sg{i}",
            onenoteSection_id=f"section{i}",
            onenotePage_id=f"page{i}",
        )
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*tasks)
    for i, result in enumerate(results):
        if i % 10 == 0:
            pass
        else:
            pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_copy_to_section_throughput_small_load():
    """Throughput: Test with small load of concurrent requests."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    tasks = [
        datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id="nb",
            sectionGroup_id="sg",
            onenoteSection_id="sec",
            onenotePage_id=f"page{i}",
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_copy_to_section_throughput_medium_load():
    """Throughput: Test with medium load of concurrent requests."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    tasks = [
        datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id="nb",
            sectionGroup_id="sg",
            onenoteSection_id="sec",
            onenotePage_id=f"page{i}",
        )
        for i in range(20)
    ]
    results = await asyncio.gather(*tasks)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_copy_to_section_throughput_high_volume():
    """Throughput: Test with high volume of concurrent requests (up to 200)."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    tasks = [
        datasource.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id="nb",
            sectionGroup_id="sg",
            onenoteSection_id="sec",
            onenotePage_id=f"page{i}",
        )
        for i in range(200)
    ]
    results = await asyncio.gather(*tasks)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_copy_to_section_throughput_mixed_success_failure():
    """Throughput: Test with mixed success and failure responses under load."""
    datasources = [
        OneNoteDataSource(DummyMSGraphClient(should_fail=(i % 25 == 0)))
        for i in range(100)
    ]
    tasks = [
        ds.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section(
            notebook_id="nb",
            sectionGroup_id="sg",
            onenoteSection_id="sec",
            onenotePage_id=f"page{i}",
        )
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*tasks)
    for i, result in enumerate(results):
        if i % 25 == 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.me_onenote_notebooks_notebook_section_groups_section_group_sections_onenote_section_pages_onenote_page_copy_to_section-mjc5mm33 and push.