⚡️ Speed up method OneNoteDataSource.me_onenote_notebooks_sections_create_pages by 25%

[codeflash-ai]

📄 25% (0.25x) speedup for OneNoteDataSource.me_onenote_notebooks_sections_create_pages in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 957 microseconds → 769 microseconds (best of 83 runs)

📝 Explanation and details

The optimized code achieves a 24% runtime improvement by eliminating unnecessary object instantiations when no query parameters are provided.

Key optimization: Conditional object creation

• Original: Always creates RequestConfiguration() objects for both query_params and config, even when no parameters are set
• Optimized: Only creates these objects when actually needed (when parameters are provided)

Specific changes:

1. Query parameters object: Only instantiates RequestConfiguration() for query_params if any of the optional parameters (select, expand, filter, etc.) are provided
2. Main config object: Only creates the main RequestConfiguration() if there are query parameters, headers, or search functionality needed
3. Passes None instead of empty objects: When no configuration is needed, passes None to the API call rather than an empty configuration object

Why this improves performance:

• Reduces object allocation overhead: Python object creation and initialization has measurable cost, especially for complex objects like RequestConfiguration
• Eliminates unnecessary attribute assignments: Avoids setting properties on objects that won't be used
• Reduces memory pressure: Fewer temporary objects created and garbage collected

Performance impact analysis:
The line profiler shows the optimization is most effective for calls with minimal or no query parameters - common in basic OneNote page creation scenarios. The test results demonstrate consistent improvements across different load patterns, with the optimization being particularly beneficial for:

• Basic page creation without complex query parameters (most common use case)
• High-volume concurrent operations where the overhead reduction compounds
• Throughput scenarios where reduced per-operation overhead translates to higher operations per second

This optimization maintains identical functionality while reducing computational overhead in the parameter-handling logic that executes on every API call.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # Used for running async functions and concurrency
# Stub for the logger

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


# Stub for MSGraphClient and its chained methods
class FakePagesPost:
    def __init__(self, response):
        self._response = response

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


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

    def by_onenote_section_id(self, onenoteSection_id):
        return self

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


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

    def by_notebook_id(self, notebook_id):
        return FakeSections(self._response)


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

    @property
    def onenote(self):
        return self

    @property
    def notebooks(self):
        return FakeNotebooks(self._response)


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

    @property
    def me(self):
        return FakeMe(self._response)


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

    def get_client(self):
        return self

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


# ---- UNIT TESTS ----

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_create_pages_basic_success():
    """Test basic successful creation of a page."""
    # Simulate a successful OneNote API response
    fake_response = {"id": "page123", "title": "Test Page"}
    client = FakeMSGraphClient(fake_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook1",
        onenoteSection_id="sectionA",
        request_body={"title": "Test Page"},
    )


@pytest.mark.asyncio
async def test_create_pages_basic_error_response_dict():
    """Test error handling with error in response dict."""
    fake_response = {"error": {"code": "BadRequest", "message": "Invalid section"}}
    client = FakeMSGraphClient(fake_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook1", onenoteSection_id="badsection"
    )


@pytest.mark.asyncio
async def test_create_pages_basic_error_response_attr():
    """Test error handling with error as attribute."""

    class ErrorObj:
        error = "Something went wrong"

    fake_response = ErrorObj()
    client = FakeMSGraphClient(fake_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook1", onenoteSection_id="sectionA"
    )


@pytest.mark.asyncio
async def test_create_pages_basic_none_response():
    """Test handling of None response from API."""
    fake_response = None
    client = FakeMSGraphClient(fake_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook1", onenoteSection_id="sectionA"
    )


@pytest.mark.asyncio
async def test_create_pages_basic_select_expand_options():
    """Test select and expand query parameters are handled."""
    fake_response = {"id": "page456", "title": "Expanded Page"}
    client = FakeMSGraphClient(fake_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook2",
        onenoteSection_id="sectionB",
        select=["id", "title"],
        expand=["parentNotebook"],
        request_body={"title": "Expanded Page"},
    )


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_create_pages_error_code_message_attrs():
    """Test error handling with code and message attributes."""

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

    fake_response = ErrorObj()
    client = FakeMSGraphClient(fake_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook3", onenoteSection_id="sectionC"
    )


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

    class BadClient:
        def get_client(self):
            class Dummy:
                def get_ms_graph_service_client(self):
                    return object()  # No 'me' property

            return Dummy()

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


@pytest.mark.asyncio
async def test_create_pages_exception_in_post():
    """Test handling exception during async post call."""

    class FailingPagesPost:
        async def post(self, body=None, request_configuration=None):
            raise RuntimeError("API failure")

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

        @property
        def pages(self):
            return FailingPagesPost()

    class FailingNotebooks:
        def by_notebook_id(self, notebook_id):
            return FailingSections()

    class FailingMe:
        @property
        def onenote(self):
            return self

        @property
        def notebooks(self):
            return FailingNotebooks()

    class FailingClient:
        @property
        def me(self):
            return FailingMe()

    class FailingMSGraphClient:
        def get_client(self):
            return self

        def get_ms_graph_service_client(self):
            return FailingClient()

    ds = OneNoteDataSource(FailingMSGraphClient())
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebookX", onenoteSection_id="sectionY"
    )


@pytest.mark.asyncio
async def test_create_pages_concurrent_execution():
    """Test concurrent execution of multiple page creation requests."""
    fake_response1 = {"id": "page1"}
    fake_response2 = {"id": "page2"}
    client1 = FakeMSGraphClient(fake_response1)
    client2 = FakeMSGraphClient(fake_response2)
    ds1 = OneNoteDataSource(client1)
    ds2 = OneNoteDataSource(client2)
    # Run two requests concurrently
    results = await asyncio.gather(
        ds1.me_onenote_notebooks_sections_create_pages(
            notebook_id="notebookA",
            onenoteSection_id="sectionA",
            request_body={"title": "Page 1"},
        ),
        ds2.me_onenote_notebooks_sections_create_pages(
            notebook_id="notebookB",
            onenoteSection_id="sectionB",
            request_body={"title": "Page 2"},
        ),
    )


@pytest.mark.asyncio
async def test_create_pages_search_consistency_header():
    """Test that ConsistencyLevel header is set when search parameter is used."""
    fake_response = {"id": "page_search"}
    client = FakeMSGraphClient(fake_response)
    ds = OneNoteDataSource(client)
    # We can't directly inspect headers, but we can check that the request succeeds
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebookS", onenoteSection_id="sectionS", search="important"
    )


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_create_pages_large_scale_concurrent():
    """Test concurrent creation of many pages (scalability)."""
    num_requests = 20  # Reasonable for unit test
    responses = [{"id": f"page{i}"} for i in range(num_requests)]
    clients = [FakeMSGraphClient(resp) for resp in responses]
    datasources = [OneNoteDataSource(c) for c in clients]
    coros = [
        ds.me_onenote_notebooks_sections_create_pages(
            notebook_id=f"notebook{i}",
            onenoteSection_id=f"section{i}",
            request_body={"title": f"Page {i}"},
        )
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_create_pages_large_scale_with_varied_params():
    """Test concurrent creation with varied query params."""
    num_requests = 10
    responses = [{"id": f"page{i}", "title": f"title{i}"} for i in range(num_requests)]
    clients = [FakeMSGraphClient(resp) for resp in responses]
    datasources = [OneNoteDataSource(c) for c in clients]
    coros = [
        ds.me_onenote_notebooks_sections_create_pages(
            notebook_id=f"notebook{i}",
            onenoteSection_id=f"section{i}",
            select=["id", "title"],
            expand=["parentNotebook"],
            filter=f"title eq 'title{i}'",
            orderby="title",
            top=1,
            skip=0,
            request_body={"title": f"title{i}"},
        )
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_create_pages_throughput_small_load():
    """Throughput: test small batch of requests."""
    num_requests = 5
    responses = [{"id": f"tp_page{i}"} for i in range(num_requests)]
    clients = [FakeMSGraphClient(resp) for resp in responses]
    datasources = [OneNoteDataSource(c) for c in clients]
    coros = [
        ds.me_onenote_notebooks_sections_create_pages(
            notebook_id=f"tp_notebook{i}",
            onenoteSection_id=f"tp_section{i}",
            request_body={"title": f"TP {i}"},
        )
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_create_pages_throughput_medium_load():
    """Throughput: test medium batch of requests."""
    num_requests = 30
    responses = [{"id": f"tp_page{i}"} for i in range(num_requests)]
    clients = [FakeMSGraphClient(resp) for resp in responses]
    datasources = [OneNoteDataSource(c) for c in clients]
    coros = [
        ds.me_onenote_notebooks_sections_create_pages(
            notebook_id=f"tp_notebook{i}",
            onenoteSection_id=f"tp_section{i}",
            request_body={"title": f"TP {i}"},
        )
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_create_pages_throughput_high_volume():
    """Throughput: test high volume batch of requests."""
    num_requests = 100
    responses = [{"id": f"tp_page{i}"} for i in range(num_requests)]
    clients = [FakeMSGraphClient(resp) for resp in responses]
    datasources = [OneNoteDataSource(c) for c in clients]
    coros = [
        ds.me_onenote_notebooks_sections_create_pages(
            notebook_id=f"tp_notebook{i}",
            onenoteSection_id=f"tp_section{i}",
            request_body={"title": f"TP {i}"},
        )
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*coros)
    for i, result 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
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 and response objects ---


class OneNoteResponse:
    """Simple response wrapper for OneNote operations."""

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


class DummyPages:
    """Stub for pages endpoint."""

    def __init__(self, notebook_id, section_id):
        self.notebook_id = notebook_id
        self.section_id = section_id
        self._responses = {}

    async def post(self, body=None, request_configuration=None):
        # Simulate a successful creation with echo of input
        if body is not None and "fail" in body:
            raise Exception("Simulated failure")
        # Simulate a realistic response object
        return {
            "id": "page123",
            "body": body,
            "notebook_id": self.notebook_id,
            "section_id": self.section_id,
        }


class DummySections:
    """Stub for sections endpoint."""

    def __init__(self, notebook_id):
        self.notebook_id = notebook_id

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


class DummyNotebooks:
    """Stub for notebooks endpoint."""

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


class DummyMe:
    """Stub for me endpoint."""

    def __init__(self):
        self.onenote = DummyOnenote()


class DummyOnenote:
    """Stub for onenote endpoint."""

    def __init__(self):
        self.notebooks = DummyNotebooks()


class DummyClient:
    """Stub for Graph client."""

    def __init__(self):
        self.me = DummyMe()

    def get_ms_graph_service_client(self):
        return self


class DummyMSGraphClient:
    """Stub for MSGraphClient wrapper."""

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

    def get_client(self):
        return self._client


# --- Unit Tests ---


@pytest.mark.asyncio
async def test_basic_successful_creation():
    """Test basic successful creation of a page with required parameters."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook1",
        onenoteSection_id="sectionA",
        request_body={"title": "Test Page", "content": "Hello World"},
    )


@pytest.mark.asyncio
async def test_basic_with_optional_parameters():
    """Test creation with all optional parameters provided."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook2",
        onenoteSection_id="sectionB",
        select=["id", "title"],
        expand=["parentNotebook"],
        filter="title eq 'Test'",
        orderby="title",
        search="test",
        top=5,
        skip=1,
        request_body={"title": "Another Page"},
        headers={"Custom-Header": "value"},
    )


@pytest.mark.asyncio
async def test_basic_empty_request_body():
    """Test creation with no request body."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook3", onenoteSection_id="sectionC", request_body=None
    )


@pytest.mark.asyncio
async def test_edge_missing_notebook_id():
    """Test with missing notebook_id (should raise TypeError)."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    with pytest.raises(TypeError):
        await ds.me_onenote_notebooks_sections_create_pages(
            onenoteSection_id="sectionD", request_body={"title": "Missing Notebook"}
        )


@pytest.mark.asyncio
async def test_edge_missing_section_id():
    """Test with missing onenoteSection_id (should raise TypeError)."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    with pytest.raises(TypeError):
        await ds.me_onenote_notebooks_sections_create_pages(
            notebook_id="notebook4", request_body={"title": "Missing Section"}
        )


@pytest.mark.asyncio
async def test_edge_invalid_request_body_type():
    """Test with invalid request_body type (should still succeed as function does not enforce type)."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook5", onenoteSection_id="sectionE", request_body="not a dict"
    )


@pytest.mark.asyncio
async def test_edge_api_exception_handling():
    """Test that exceptions in the API call are handled gracefully."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook6",
        onenoteSection_id="sectionF",
        request_body={"fail": True},
    )


@pytest.mark.asyncio
async def test_edge_none_response_handling():
    """Test that None response is handled as an error."""

    class DummyPagesNone(DummyPages):
        async def post(self, body=None, request_configuration=None):
            return None

    class DummySectionsNone(DummySections):
        def by_onenote_section_id(self, section_id):
            return DummyPagesNone(self.notebook_id, section_id)

    class DummyNotebooksNone(DummyNotebooks):
        def by_notebook_id(self, notebook_id):
            return DummySectionsNone(notebook_id)

    class DummyOnenoteNone(DummyOnenote):
        def __init__(self):
            self.notebooks = DummyNotebooksNone()

    class DummyMeNone(DummyMe):
        def __init__(self):
            self.onenote = DummyOnenoteNone()

    class DummyClientNone(DummyClient):
        def __init__(self):
            self.me = DummyMeNone()

        def get_ms_graph_service_client(self):
            return self

    class DummyMSGraphClientNone(DummyMSGraphClient):
        def __init__(self):
            self._client = DummyClientNone()

        def get_client(self):
            return self._client

    ds = OneNoteDataSource(DummyMSGraphClientNone())
    result = await ds.me_onenote_notebooks_sections_create_pages(
        notebook_id="notebook7",
        onenoteSection_id="sectionG",
        request_body={"title": "None response"},
    )


@pytest.mark.asyncio
async def test_concurrent_execution():
    """Test concurrent execution of multiple page creations."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    coros = [
        ds.me_onenote_notebooks_sections_create_pages(
            notebook_id=f"notebook{i}",
            onenoteSection_id=f"section{i}",
            request_body={"title": f"Page {i}"},
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_large_scale_concurrent_creation():
    """Test large scale concurrent creation (up to 50 pages)."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    coros = [
        ds.me_onenote_notebooks_sections_create_pages(
            notebook_id=f"notebook{i}",
            onenoteSection_id=f"section{i}",
            request_body={"title": f"Bulk Page {i}"},
        )
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_throughput_small_load():
    """Throughput test: create 5 pages concurrently and check all succeed quickly."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    coros = [
        ds.me_onenote_notebooks_sections_create_pages(
            notebook_id="notebookT",
            onenoteSection_id="sectionT",
            request_body={"title": f"TP Small {i}"},
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_throughput_medium_load():
    """Throughput test: create 25 pages concurrently and check all succeed."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    coros = [
        ds.me_onenote_notebooks_sections_create_pages(
            notebook_id="notebookM",
            onenoteSection_id="sectionM",
            request_body={"title": f"TP Medium {i}"},
        )
        for i in range(25)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_notebooks_sections_create_pages_throughput_high_load():
    """Throughput test: create 100 pages concurrently and check all succeed."""
    ds = OneNoteDataSource(DummyMSGraphClient())
    coros = [
        ds.me_onenote_notebooks_sections_create_pages(
            notebook_id="notebookH",
            onenoteSection_id="sectionH",
            request_body={"title": f"TP High {i}"},
        )
        for i in range(100)
    ]
    results = await asyncio.gather(*coros)


# 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_sections_create_pages-mjcgk98w and push.