⚡️ Speed up method OneNoteDataSource.me_onenote_update_notebooks by 9%

[codeflash-ai]

📄 9% (0.09x) speedup for OneNoteDataSource.me_onenote_update_notebooks in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 1.03 milliseconds → 944 microseconds (best of 73 runs)

📝 Explanation and details

The optimized code achieves a 9% runtime improvement through two key optimizations focused on eliminating redundant operations and reordering checks for better performance:

1. Response Handler Optimization (_handle_onenote_response)

• Early returns: The original code checked all error conditions before constructing the response, but the optimized version returns immediately when an error is found, eliminating unnecessary subsequent checks
• Reordered checks: Moved isinstance(response, dict) check before hasattr() calls since dictionary type checking and key lookup ('error' in response) is faster than attribute reflection via hasattr()
• Reduced variable assignments: Eliminated the intermediate success and error_msg variables for the success path, directly constructing the final response

2. Request Configuration Optimization (me_onenote_update_notebooks)

• Eliminated double instantiation: The original code created two RequestConfiguration objects (query_params and config), but the optimized version reuses a single config object, reducing memory allocation overhead
• Pre-computed list conversions: Moved the select/expand list conversion logic outside the conditional blocks to avoid repeated type checking and list creation
• Defensive copying: Added .copy() when assigning headers to prevent mutation of the input reference while maintaining the same behavior

Performance Impact Analysis:
The line profiler shows the response handler improved from 1.47ms to 0.99ms (32% faster), while the main function improved from 6.51ms to 5.44ms (16% faster). These optimizations are particularly effective for high-frequency OneNote API operations since they reduce the per-call overhead of object creation and attribute lookups.

Test Case Benefits:
The optimizations show consistent improvements across all test scenarios, with the most benefit in concurrent execution tests (large-scale and throughput tests) where the reduced object allocation overhead compounds across multiple simultaneous operations.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 601 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
# Patch RequestConfiguration in the tested module to our dummy version

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

# --- Minimal stubs for dependencies ---


class OneNoteResponse:
    def __init__(self, success, data=None, error=None):
        self.success = success
        self.data = data
        self.error = error


# --- Helper: Dummy client hierarchy for patch call mocking ---


class DummyPatch:
    def __init__(self, response_to_return=None, raise_exc=None):
        self.response_to_return = response_to_return
        self.raise_exc = raise_exc
        self.called_with = []

    async def patch(self, body=None, request_configuration=None):
        self.called_with.append((body, request_configuration))
        if self.raise_exc:
            raise self.raise_exc
        return self.response_to_return


class DummyByNotebookId:
    def __init__(self, patcher):
        self._patcher = patcher

    def patch(self, body=None, request_configuration=None):
        return self._patcher.patch(
            body=body, request_configuration=request_configuration
        )


class DummyNotebooks:
    def __init__(self, patcher):
        self._patcher = patcher

    def by_notebook_id(self, notebook_id):
        return DummyByNotebookId(self._patcher)


class DummyOneNote:
    def __init__(self, patcher):
        self.notebooks = DummyNotebooks(patcher)


class DummyMe:
    def __init__(self, patcher):
        self.onenote = DummyOneNote(patcher)


class DummyGraphClient:
    def __init__(self, patcher):
        self.me = DummyMe(patcher)

    def get_ms_graph_service_client(self):
        return self


class DummyClientWrapper:
    def __init__(self, patcher):
        self._patcher = patcher

    def get_client(self):
        return DummyGraphClient(self._patcher)


# --- TESTS ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_success_basic():
    """Test basic successful update with minimal arguments."""
    dummy_response = {"id": "notebook123", "name": "Updated Notebook"}
    patcher = DummyPatch(response_to_return=dummy_response)
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_update_notebooks("notebook123")


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_with_all_parameters():
    """Test update with all parameters set."""
    dummy_response = {"id": "notebook456", "name": "Notebook456"}
    patcher = DummyPatch(response_to_return=dummy_response)
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_update_notebooks(
        "notebook456",
        select=["id", "name"],
        expand=["sections"],
        filter="name eq 'Notebook456'",
        orderby="name desc",
        search="Notebook",
        top=5,
        skip=1,
        request_body={"name": "Notebook456"},
        headers={"Custom-Header": "Value"},
    )


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_async_behavior():
    """Test that the function is a coroutine and can be awaited."""
    dummy_response = {"id": "n1"}
    patcher = DummyPatch(response_to_return=dummy_response)
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    # Ensure the function returns a coroutine
    codeflash_output = ds.me_onenote_update_notebooks("n1")
    coro = codeflash_output
    result = await coro


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_handles_exception_in_patch():
    """Test that exceptions in patch are caught and wrapped in OneNoteResponse."""
    patcher = DummyPatch(raise_exc=RuntimeError("Patch failed"))
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_update_notebooks("notebook999")


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_handles_none_response():
    """Test that None response from patch is handled as an error."""
    patcher = DummyPatch(response_to_return=None)
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_update_notebooks("notebook_none")


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_handles_dict_error_response():
    """Test that a dict with 'error' key is handled as error."""
    dummy_response = {"error": {"code": "Invalid", "message": "Bad notebook"}}
    patcher = DummyPatch(response_to_return=dummy_response)
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_update_notebooks("notebook_err")


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

    class ErrorObj:
        error = "Not found"

    patcher = DummyPatch(response_to_return=ErrorObj())
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_update_notebooks("notebook_obj_err")


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_handles_code_message_response():
    """Test that an object with 'code' and 'message' is handled as error."""

    class CodeMsgObj:
        code = "404"
        message = "Notebook not found"

    patcher = DummyPatch(response_to_return=CodeMsgObj())
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_update_notebooks("notebook_code_msg")


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_concurrent_execution():
    """Test concurrent calls to ensure async safety."""
    dummy_response = {"id": "concurrent", "name": "Concurrent"}
    patcher = DummyPatch(response_to_return=dummy_response)
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    coros = [
        ds.me_onenote_update_notebooks(f"notebook_{i}", request_body={"name": f"n{i}"})
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_large_scale_concurrent():
    """Test 50 concurrent calls to check scalability and async correctness."""
    dummy_response = {"id": "large", "name": "Large"}
    patcher = DummyPatch(response_to_return=dummy_response)
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    coros = [
        ds.me_onenote_update_notebooks(f"notebook_{i}", request_body={"name": f"n{i}"})
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_large_scale_error_handling():
    """Test that all errors are handled gracefully in concurrent error scenario."""
    patcher = DummyPatch(raise_exc=ValueError("Simultaneous error"))
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    coros = [ds.me_onenote_update_notebooks(f"notebook_{i}") for i in range(10)]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_throughput_small_load():
    """Throughput: Test 10 concurrent successful requests."""
    dummy_response = {"id": "tp_small", "name": "ThroughputSmall"}
    patcher = DummyPatch(response_to_return=dummy_response)
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    coros = [ds.me_onenote_update_notebooks(f"notebook_{i}") for i in range(10)]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_throughput_medium_load():
    """Throughput: Test 100 concurrent successful requests."""
    dummy_response = {"id": "tp_medium", "name": "ThroughputMedium"}
    patcher = DummyPatch(response_to_return=dummy_response)
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    coros = [ds.me_onenote_update_notebooks(f"notebook_{i}") for i in range(100)]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_throughput_mixed_success_error():
    """Throughput: Test mixed success and error responses under load."""

    # Alternate between success and error
    class AltPatch:
        def __init__(self):
            self.count = 0

        async def patch(self, body=None, request_configuration=None):
            if self.count % 2 == 0:
                resp = {"id": f"tp_{self.count}", "name": "OK"}
            else:
                raise RuntimeError(f"Err_{self.count}")
            self.count += 1
            return resp

    patcher = AltPatch()
    client = DummyClientWrapper(patcher)
    ds = OneNoteDataSource(client)
    coros = [ds.me_onenote_update_notebooks(f"notebook_{i}") for i in range(20)]
    results = await asyncio.gather(*coros)
    # Even indices should be success, odd should be error
    for i, result in enumerate(results):
        if i % 2 == 0:
            pass
        else:
            pass


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

import asyncio  # For running async functions and concurrency
# Patch logger used in OneNoteDataSource to avoid NameError

import pytest  # For unit testing
from app.sources.external.microsoft.one_note.one_note import OneNoteDataSource

# --- Mocks and minimal stubs for dependencies ---


class OneNoteResponse:
    """Minimal stub for OneNoteResponse used in the function."""

    def __init__(self, success: bool, data=None, error=None):
        self.success = success
        self.data = data
        self.error = error


class DummyNotebookClient:
    """Mocks the .patch() async method for the notebook resource."""

    def __init__(self, patch_result=None, raise_exc=None):
        self.patch_result = patch_result
        self.raise_exc = raise_exc
        self.calls = []

    async def patch(self, body=None, request_configuration=None):
        self.calls.append((body, request_configuration))
        if self.raise_exc:
            raise self.raise_exc
        return self.patch_result


class DummyNotebooks:
    """Mocks .by_notebook_id() to return DummyNotebookClient."""

    def __init__(self, patch_result=None, raise_exc=None):
        self.patch_result = patch_result
        self.raise_exc = raise_exc
        self.last_notebook_id = None
        self.calls = []

    def by_notebook_id(self, notebook_id):
        self.last_notebook_id = notebook_id
        self.calls.append(notebook_id)
        return DummyNotebookClient(self.patch_result, self.raise_exc)


class DummyMe:
    """Mocks .onenote.notebooks for the client."""

    def __init__(self, patch_result=None, raise_exc=None):
        self.onenote = type("Onenote", (), {})()
        self.onenote.notebooks = DummyNotebooks(patch_result, raise_exc)


class DummyClient:
    """Mocks the MSGraphClient.get_client().get_ms_graph_service_client().me chain."""

    def __init__(self, patch_result=None, raise_exc=None):
        self.me = DummyMe(patch_result, raise_exc)

    def get_ms_graph_service_client(self):
        return self

    def get_client(self):
        return self


# --- Unit Tests ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_basic_success():
    """Test basic successful update with minimal required arguments."""
    dummy_response = {"id": "notebook123", "name": "Test Notebook"}
    ds = OneNoteDataSource(client=DummyClient(patch_result=dummy_response))
    result = await ds.me_onenote_update_notebooks(
        "notebook123", request_body={"name": "Test Notebook"}
    )


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_basic_with_all_params():
    """Test update with all optional parameters provided."""
    dummy_response = {"id": "notebook456", "name": "Updated Notebook"}
    ds = OneNoteDataSource(client=DummyClient(patch_result=dummy_response))
    result = await ds.me_onenote_update_notebooks(
        notebook_id="notebook456",
        select=["id", "name"],
        expand=["sections"],
        filter="name eq 'Updated Notebook'",
        orderby="name desc",
        search="Updated",
        top=5,
        skip=2,
        request_body={"name": "Updated Notebook"},
        headers={"Custom-Header": "Value"},
    )


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_basic_async_behavior():
    """Test that the function is awaitable and returns a coroutine."""
    ds = OneNoteDataSource(client=DummyClient(patch_result={"id": "nb"}))
    codeflash_output = ds.me_onenote_update_notebooks(
        "nb", request_body={"name": "Async"}
    )
    coro = codeflash_output
    result = await coro


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_handles_none_response():
    """Test that None response from patch results in failure OneNoteResponse."""
    ds = OneNoteDataSource(client=DummyClient(patch_result=None))
    result = await ds.me_onenote_update_notebooks(
        "notebook999", request_body={"name": "None"}
    )


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_handles_error_attr():
    """Test response object with .error attribute triggers error handling."""

    class ErrorObj:
        error = "Some error"

    ds = OneNoteDataSource(client=DummyClient(patch_result=ErrorObj()))
    result = await ds.me_onenote_update_notebooks("notebook_err", request_body={})


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_handles_error_dict():
    """Test response dict with 'error' key triggers error handling."""
    error_dict = {"error": {"code": "BadRequest", "message": "Invalid notebook"}}
    ds = OneNoteDataSource(client=DummyClient(patch_result=error_dict))
    result = await ds.me_onenote_update_notebooks("notebook_bad", request_body={})


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

    class Resp:
        code = "Forbidden"
        message = "You do not have permission"

    ds = OneNoteDataSource(client=DummyClient(patch_result=Resp()))
    result = await ds.me_onenote_update_notebooks("notebook_forbid", request_body={})


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_exception_handling():
    """Test that exceptions in patch are caught and returned as error."""
    ds = OneNoteDataSource(client=DummyClient(raise_exc=RuntimeError("API failure")))
    result = await ds.me_onenote_update_notebooks("notebook_exc", request_body={})


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_concurrent_execution():
    """Test concurrent execution of multiple updates."""
    dummy_response1 = {"id": "nb1", "name": "Notebook1"}
    dummy_response2 = {"id": "nb2", "name": "Notebook2"}
    # Each call needs its own DummyClient to avoid shared state issues
    ds1 = OneNoteDataSource(client=DummyClient(patch_result=dummy_response1))
    ds2 = OneNoteDataSource(client=DummyClient(patch_result=dummy_response2))
    results = await asyncio.gather(
        ds1.me_onenote_update_notebooks("nb1", request_body={"name": "Notebook1"}),
        ds2.me_onenote_update_notebooks("nb2", request_body={"name": "Notebook2"}),
    )


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_search_consistency_level_header():
    """Test that ConsistencyLevel header is set if search param is given."""

    class HeaderCheckNotebookClient(DummyNotebookClient):
        async def patch(self, body=None, request_configuration=None):
            return {"id": "nb", "name": "Search"}

    class HeaderCheckNotebooks:
        def by_notebook_id(self, notebook_id):
            return HeaderCheckNotebookClient()

    class HeaderCheckMe:
        def __init__(self):
            self.onenote = type("Onenote", (), {})()
            self.onenote.notebooks = HeaderCheckNotebooks()

    class HeaderCheckClient:
        def get_ms_graph_service_client(self):
            return self

        def get_client(self):
            return self

        @property
        def me(self):
            return HeaderCheckMe()

    ds = OneNoteDataSource(client=HeaderCheckClient())
    result = await ds.me_onenote_update_notebooks(
        notebook_id="nb", search="query", request_body={"name": "Search"}
    )


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_many_concurrent_updates():
    """Test many concurrent updates to ensure scalability and no shared state bugs."""
    N = 20  # Large, but not excessive
    responses = [{"id": f"nb{i}", "name": f"Notebook{i}"} for i in range(N)]
    datasources = [
        OneNoteDataSource(client=DummyClient(patch_result=resp)) for resp in responses
    ]
    tasks = [
        ds.me_onenote_update_notebooks(f"nb{i}", request_body={"name": f"Notebook{i}"})
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*tasks)
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_large_payload():
    """Test update with a large request_body."""
    large_body = {"data": "x" * 10000}  # 10KB payload
    dummy_response = {"id": "large", "name": "LargePayload"}
    ds = OneNoteDataSource(client=DummyClient(patch_result=dummy_response))
    result = await ds.me_onenote_update_notebooks("large", request_body=large_body)


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_throughput_small_load():
    """Throughput: Test multiple small, fast updates in parallel."""
    N = 10
    datasources = [
        OneNoteDataSource(client=DummyClient(patch_result={"id": f"nb{i}"}))
        for i in range(N)
    ]
    tasks = [
        ds.me_onenote_update_notebooks(f"nb{i}", request_body={"name": f"NB{i}"})
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_throughput_medium_load():
    """Throughput: Test with a moderate number of concurrent updates."""
    N = 50
    datasources = [
        OneNoteDataSource(client=DummyClient(patch_result={"id": f"nb{i}"}))
        for i in range(N)
    ]
    tasks = [
        ds.me_onenote_update_notebooks(f"nb{i}", request_body={"name": f"NB{i}"})
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_me_onenote_update_notebooks_throughput_high_volume():
    """Throughput: Test with a high, but bounded, number of updates."""
    N = 100
    datasources = [
        OneNoteDataSource(client=DummyClient(patch_result={"id": f"nb{i}"}))
        for i in range(N)
    ]
    tasks = [
        ds.me_onenote_update_notebooks(f"nb{i}", request_body={"name": f"NB{i}"})
        for i, ds in enumerate(datasources)
    ]
    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.me_onenote_update_notebooks-mjbn7qi9 and push.