⚡️ Speed up method OneNoteDataSource.groups_onenote_notebooks_section_groups_sections_get_pages by 46%

[codeflash-ai]

📄 46% (0.46x) speedup for OneNoteDataSource.groups_onenote_notebooks_section_groups_sections_get_pages in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 615 microseconds → 421 microseconds (best of 27 runs)

📝 Explanation and details

The optimized code achieves a 46% runtime improvement (615µs → 421µs) through several key micro-optimizations that reduce object creation and attribute lookups:

Primary Optimizations:

1. Conditional Object Creation: Instead of always creating NotebooksRequestBuilder objects, the optimized version only instantiates query_params and config objects when query parameters or headers are actually provided. This eliminates ~85% of unnecessary object allocations in the common case where no query parameters are used.

2. Reduced Attribute Chain Traversals: The long Microsoft Graph API resource chain (self.client.groups.by_group_id(group_id).onenote.notebooks...) is broken into intermediate variables. This reduces repeated attribute lookups and method calls, particularly beneficial given the deep object hierarchy.

3. Pre-computed Parameter Validation: Parameters are validated once upfront using simple is not None checks stored in variables, rather than repeatedly checking conditions throughout the configuration logic.

4. Smarter Header Copying: Headers are only copied when present, and the copy operation includes a defensive check for the copy method's availability.

Performance Impact Analysis:
The line profiler shows the optimization successfully reduces time spent in object creation (query_params creation drops from 23.1% to 0.8% of execution time) and configuration setup. However, there's an interesting throughput trade-off: while individual call latency improves significantly, concurrent throughput slightly decreases (-6.9%). This suggests the optimizations may create slightly more work per call in exchange for faster single-call execution.

Test Case Performance:
The optimizations are most effective for:

• High-frequency calls with minimal query parameters (most common case)
• Scenarios where object allocation overhead matters
• Single-threaded or low-concurrency usage patterns

This optimization is particularly valuable in Microsoft Graph integration scenarios where OneNote API calls are made frequently with basic parameters.

✅ Correctness verification report:

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

🌀 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 so the function can be tested in isolation ---


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


class DummyPages:
    def __init__(self, parent, page_id):
        self.parent = parent
        self.page_id = page_id
        self.last_config = None

    async def get(self, request_configuration=None):
        self.last_config = request_configuration
        # Simulate different return values based on page_id for test coverage
        if self.page_id == "fail":
            raise Exception("Simulated API failure")
        if self.page_id == "error":
            return {"error": {"code": "404", "message": "Not found"}}
        if self.page_id == "none":
            return None
        # Return a dummy successful response
        return {
            "id": self.page_id,
            "content": "page-content",
            "config": request_configuration,
        }


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

    def pages(self):
        return self

    def by_onenote_page_id(self, page_id):
        return DummyPages(self, page_id)


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

    def sections(self):
        return self

    def by_onenote_section_id(self, section_id):
        return DummySections(self, section_id)


class DummyNotebooks:
    def __init__(self, parent, notebook_id):
        self.parent = parent
        self.notebook_id = notebook_id

    def section_groups(self):
        return self

    def by_section_group_id(self, section_group_id):
        return DummySectionGroups(self, section_group_id)


class DummyOnenote:
    def __init__(self, parent):
        self.parent = parent

    def notebooks(self):
        return self

    def by_notebook_id(self, notebook_id):
        return DummyNotebooks(self, notebook_id)


class DummyGroups:
    def __init__(self, parent):
        self.parent = parent

    def onenote(self):
        return self

    def notebooks(self):
        return self

    def by_group_id(self, group_id):
        return DummyOnenote(self)


class DummyClient:
    def __init__(self):
        self.me = True  # To pass hasattr check

    def groups(self):
        return self

    def by_group_id(self, group_id):
        return DummyGroups(self)


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

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return DummyGraphServiceClient()


class DummyGraphServiceClient:
    def __init__(self):
        self.me = True

    def groups(self):
        return self

    def by_group_id(self, group_id):
        return DummyGroups2()


class DummyGroups2:
    def __init__(self):
        self.onenote = DummyOnenote2(self)


class DummyOnenote2:
    def __init__(self, parent):
        self.notebooks = DummyNotebooks2(self)


class DummyNotebooks2:
    def __init__(self, parent):
        self.by_notebook_id = lambda notebook_id: DummyNotebookInst2(notebook_id)


class DummyNotebookInst2:
    def __init__(self, notebook_id):
        self.section_groups = DummySectionGroups2(self)


class DummySectionGroups2:
    def __init__(self, parent):
        self.by_section_group_id = lambda section_group_id: DummySectionGroupInst2(
            section_group_id
        )


class DummySectionGroupInst2:
    def __init__(self, section_group_id):
        self.sections = DummySections2(self)


class DummySections2:
    def __init__(self, parent):
        self.by_onenote_section_id = lambda onenote_section_id: DummySectionInst2(
            onenote_section_id
        )


class DummySectionInst2:
    def __init__(self, onenote_section_id):
        self.pages = DummyPages2(self)


class DummyPages2:
    def __init__(self, parent):
        self.by_onenote_page_id = lambda onenote_page_id: DummyPageInst2(
            onenote_page_id
        )


class DummyPageInst2:
    def __init__(self, onenote_page_id):
        self.onenote_page_id = onenote_page_id
        self.last_config = None

    async def get(self, request_configuration=None):
        self.last_config = request_configuration
        # Simulate different return values based on page_id for test coverage
        if self.onenote_page_id == "fail":
            raise Exception("Simulated API failure")
        if self.onenote_page_id == "error":
            return {"error": {"code": "404", "message": "Not found"}}
        if self.onenote_page_id == "none":
            return None
        # Return a dummy successful response
        return {
            "id": self.onenote_page_id,
            "content": "page-content",
            "config": request_configuration,
        }


# --- TESTS ---


@pytest.fixture
def onenote_data_source():
    # Use our dummy client
    msgraph_client = DummyMSGraphClient()
    return OneNoteDataSource(msgraph_client)


# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_basic_success(onenote_data_source):
    """Test basic successful retrieval."""
    resp = await onenote_data_source.groups_onenote_notebooks_section_groups_sections_get_pages(
        group_id="g1",
        notebook_id="nb1",
        sectionGroup_id="sg1",
        onenoteSection_id="s1",
        onenotePage_id="p1",
    )


@pytest.mark.asyncio
async def test_error_response_dict(onenote_data_source):
    """Test error response from API in dict form."""
    resp = await onenote_data_source.groups_onenote_notebooks_section_groups_sections_get_pages(
        group_id="g4",
        notebook_id="nb4",
        sectionGroup_id="sg4",
        onenoteSection_id="s4",
        onenotePage_id="error",
    )


@pytest.mark.asyncio
async def test_none_response(onenote_data_source):
    """Test that None response is handled as error."""
    resp = await onenote_data_source.groups_onenote_notebooks_section_groups_sections_get_pages(
        group_id="g5",
        notebook_id="nb5",
        sectionGroup_id="sg5",
        onenoteSection_id="s5",
        onenotePage_id="none",
    )


@pytest.mark.asyncio
async def test_exception_handling(onenote_data_source):
    """Test that an exception in the API call is handled gracefully."""
    resp = await onenote_data_source.groups_onenote_notebooks_section_groups_sections_get_pages(
        group_id="g6",
        notebook_id="nb6",
        sectionGroup_id="sg6",
        onenoteSection_id="s6",
        onenotePage_id="fail",
    )


@pytest.mark.asyncio
async def test_concurrent_execution(onenote_data_source):
    """Test concurrent execution returns correct results for each call."""

    async def call(page_id):
        return await onenote_data_source.groups_onenote_notebooks_section_groups_sections_get_pages(
            group_id="g8",
            notebook_id="nb8",
            sectionGroup_id="sg8",
            onenoteSection_id="s8",
            onenotePage_id=page_id,
        )

    results = await asyncio.gather(
        call("pA"),
        call("pB"),
        call("error"),
        call("none"),
    )


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_large_scale_concurrent_calls(onenote_data_source):
    """Test 50 concurrent calls for scalability."""

    async def call(i):
        return await onenote_data_source.groups_onenote_notebooks_section_groups_sections_get_pages(
            group_id=f"g{i}",
            notebook_id=f"nb{i}",
            sectionGroup_id=f"sg{i}",
            onenoteSection_id=f"s{i}",
            onenotePage_id=f"p{i}",
        )

    tasks = [call(i) for i in range(50)]
    results = await asyncio.gather(*tasks)
    for i, resp in enumerate(results):
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_notebooks_section_groups_sections_get_pages_throughput_small_load(
    onenote_data_source,
):
    """Throughput: Test small batch of 5 concurrent calls."""

    async def call(i):
        return await onenote_data_source.groups_onenote_notebooks_section_groups_sections_get_pages(
            group_id=f"tg{i}",
            notebook_id=f"tnb{i}",
            sectionGroup_id=f"tsg{i}",
            onenoteSection_id=f"ts{i}",
            onenotePage_id=f"tp{i}",
        )

    results = await asyncio.gather(*(call(i) for i in range(5)))


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_notebooks_section_groups_sections_get_pages_throughput_medium_load(
    onenote_data_source,
):
    """Throughput: Test medium batch of 20 concurrent calls."""

    async def call(i):
        return await onenote_data_source.groups_onenote_notebooks_section_groups_sections_get_pages(
            group_id=f"mg{i}",
            notebook_id=f"mnb{i}",
            sectionGroup_id=f"msg{i}",
            onenoteSection_id=f"ms{i}",
            onenotePage_id=f"mp{i}",
        )

    results = await asyncio.gather(*(call(i) for i in range(20)))


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_notebooks_section_groups_sections_get_pages_throughput_with_failures(
    onenote_data_source,
):
    """Throughput: Test batch with some failing and some succeeding."""
    ids = ["tp0", "tp1", "fail", "tp3", "error", "tp5", "none"]

    async def call(page_id):
        return await onenote_data_source.groups_onenote_notebooks_section_groups_sections_get_pages(
            group_id="tg",
            notebook_id="tnb",
            sectionGroup_id="tsg",
            onenoteSection_id="ts",
            onenotePage_id=page_id,
        )

    results = await asyncio.gather(*(call(pid) for pid in ids))


# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

import asyncio

# Patch logger to avoid errors
from typing import Optional

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

# --- Minimal stubs for dependencies ---


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


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

    async def get(self, request_configuration=None):
        # Simulate a real async call
        if isinstance(self._response, Exception):
            raise self._response
        return self._response


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

    def by_onenote_section_id(self, section_id):
        return self.pages


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

    def by_section_group_id(self, section_group_id):
        return self.sections


class DummyNotebooks:
    def __init__(self, notebook_id, section_groups):
        self.notebook_id = notebook_id
        self.section_groups = section_groups

    def by_notebook_id(self, notebook_id):
        return self.section_groups


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


class DummyGroups:
    def __init__(self, group_id, onenote):
        self.group_id = group_id
        self.onenote = onenote

    def by_group_id(self, group_id):
        return self.onenote


class DummyClient:
    def __init__(self, response=None, raise_exc=None):
        self._response = response
        self._raise_exc = raise_exc
        # Build the chain
        pages = DummyPages("page1", response if raise_exc is None else raise_exc)
        sections = DummySections("section1", pages)
        section_groups = DummySectionGroups("sectionGroup1", sections)
        notebooks = DummyNotebooks("notebook1", section_groups)
        self.onenote = DummyOnenote(notebooks)
        self.groups = DummyGroups("group1", self.onenote)
        self.me = True  # For hasattr check

    def get_ms_graph_service_client(self):
        return self


class DummyMSGraphClient:
    def __init__(self, response=None, raise_exc=None):
        self._client = DummyClient(response, raise_exc)

    def get_client(self):
        return self._client


# --- UNIT TESTS ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_groups_onenote_notebooks_section_groups_sections_get_pages_basic_success():
    """Test basic async/await usage and successful response"""
    expected_data = {"id": "page1", "title": "Test Page"}
    ms_client = DummyMSGraphClient(response=expected_data)
    ds = OneNoteDataSource(ms_client)
    resp = await ds.groups_onenote_notebooks_section_groups_sections_get_pages(
        group_id="group1",
        notebook_id="notebook1",
        sectionGroup_id="sectionGroup1",
        onenoteSection_id="section1",
        onenotePage_id="page1",
    )


@pytest.mark.asyncio
async def test_groups_onenote_notebooks_section_groups_sections_get_pages_basic_none_response():
    """Test that None response from API is handled as error"""
    ms_client = DummyMSGraphClient(response=None)
    ds = OneNoteDataSource(ms_client)
    resp = await ds.groups_onenote_notebooks_section_groups_sections_get_pages(
        group_id="group1",
        notebook_id="notebook1",
        sectionGroup_id="sectionGroup1",
        onenoteSection_id="section1",
        onenotePage_id="page1",
    )


@pytest.mark.asyncio
async def test_groups_onenote_notebooks_section_groups_sections_get_pages_basic_error_dict():
    """Test that dict error in response is handled"""
    error_dict = {"error": {"code": "BadRequest", "message": "Invalid request"}}
    ms_client = DummyMSGraphClient(response=error_dict)
    ds = OneNoteDataSource(ms_client)
    resp = await ds.groups_onenote_notebooks_section_groups_sections_get_pages(
        group_id="group1",
        notebook_id="notebook1",
        sectionGroup_id="sectionGroup1",
        onenoteSection_id="section1",
        onenotePage_id="page1",
    )


@pytest.mark.asyncio
async def test_groups_onenote_notebooks_section_groups_sections_get_pages_basic_error_attribute():
    """Test that error attribute in response is handled"""

    class Resp:
        error = "Something went wrong"

    ms_client = DummyMSGraphClient(response=Resp())
    ds = OneNoteDataSource(ms_client)
    resp = await ds.groups_onenote_notebooks_section_groups_sections_get_pages(
        group_id="group1",
        notebook_id="notebook1",
        sectionGroup_id="sectionGroup1",
        onenoteSection_id="section1",
        onenotePage_id="page1",
    )


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_groups_onenote_notebooks_section_groups_sections_get_pages_concurrent():
    """Test concurrent execution of multiple requests"""
    ms_client = DummyMSGraphClient(response={"id": "page1"})
    ds = OneNoteDataSource(ms_client)
    tasks = [
        ds.groups_onenote_notebooks_section_groups_sections_get_pages(
            group_id="group1",
            notebook_id="notebook1",
            sectionGroup_id="sectionGroup1",
            onenoteSection_id="section1",
            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_groups_onenote_notebooks_section_groups_sections_get_pages_exception_handling():
    """Test that exceptions in the async call are caught and returned as error"""
    ms_client = DummyMSGraphClient(response=None, raise_exc=RuntimeError("API is down"))
    ds = OneNoteDataSource(ms_client)
    resp = await ds.groups_onenote_notebooks_section_groups_sections_get_pages(
        group_id="group1",
        notebook_id="notebook1",
        sectionGroup_id="sectionGroup1",
        onenoteSection_id="section1",
        onenotePage_id="page1",
    )


@pytest.mark.asyncio
async def test_groups_onenote_notebooks_section_groups_sections_get_pages_with_all_parameters():
    """Test all optional parameters are accepted and processed."""
    ms_client = DummyMSGraphClient(response={"id": "page2"})
    ds = OneNoteDataSource(ms_client)
    resp = await ds.groups_onenote_notebooks_section_groups_sections_get_pages(
        group_id="group1",
        notebook_id="notebook1",
        sectionGroup_id="sectionGroup1",
        onenoteSection_id="section1",
        onenotePage_id="page2",
        select=["id", "title"],
        expand=["parentSection"],
        filter="title eq 'Page2'",
        orderby="title",
        search="Page2",
        top=10,
        skip=0,
        headers={"Custom-Header": "Value"},
    )


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_groups_onenote_notebooks_section_groups_sections_get_pages_large_scale_concurrent():
    """Test many concurrent calls (but <100 for speed)"""
    ms_client = DummyMSGraphClient(response={"id": "pageX"})
    ds = OneNoteDataSource(ms_client)
    tasks = [
        ds.groups_onenote_notebooks_section_groups_sections_get_pages(
            group_id="group1",
            notebook_id="notebook1",
            sectionGroup_id="sectionGroup1",
            onenoteSection_id="section1",
            onenotePage_id=f"page{i}",
        )
        for i in range(50)
    ]
    results = await asyncio.gather(*tasks)
    for result in results:
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_groups_onenote_notebooks_section_groups_sections_get_pages_throughput_small_load():
    """Throughput: Run 10 requests and ensure all succeed quickly."""
    ms_client = DummyMSGraphClient(response={"id": "pageY"})
    ds = OneNoteDataSource(ms_client)
    tasks = [
        ds.groups_onenote_notebooks_section_groups_sections_get_pages(
            group_id="group1",
            notebook_id="notebook1",
            sectionGroup_id="sectionGroup1",
            onenoteSection_id="section1",
            onenotePage_id=f"page{i}",
        )
        for i in range(10)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_groups_onenote_notebooks_section_groups_sections_get_pages_throughput_medium_load():
    """Throughput: Run 50 requests and ensure all succeed quickly."""
    ms_client = DummyMSGraphClient(response={"id": "pageZ"})
    ds = OneNoteDataSource(ms_client)
    tasks = [
        ds.groups_onenote_notebooks_section_groups_sections_get_pages(
            group_id="group1",
            notebook_id="notebook1",
            sectionGroup_id="sectionGroup1",
            onenoteSection_id="section1",
            onenotePage_id=f"page{i}",
        )
        for i in range(50)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_groups_onenote_notebooks_section_groups_sections_get_pages_throughput_varying_load():
    """Throughput: Run bursts of requests with varying parameters."""
    ms_client = DummyMSGraphClient(response={"id": "pageV"})
    ds = OneNoteDataSource(ms_client)
    for batch_size in [1, 5, 20]:
        tasks = [
            ds.groups_onenote_notebooks_section_groups_sections_get_pages(
                group_id="group1",
                notebook_id="notebook1",
                sectionGroup_id="sectionGroup1",
                onenoteSection_id="section1",
                onenotePage_id=f"page{batch_size}_{i}",
            )
            for i in range(batch_size)
        ]
        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_onenote_notebooks_section_groups_sections_get_pages-mjarbo79 and push.