⚡️ Speed up method OneNoteDataSource.me_onenote_delete_notebooks by 13%

[codeflash-ai]

📄 13% (0.13x) speedup for OneNoteDataSource.me_onenote_delete_notebooks in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 29.3 milliseconds → 26.0 milliseconds (best of 174 runs)

📝 Explanation and details

The optimization achieves a 12% runtime improvement by reducing unnecessary object creation and streamlining control flow in the OneNote deletion method.

Key optimizations applied:

1. Conditional object creation: The optimized code only creates RequestConfiguration objects when actually needed, using has_params and has_custom_headers flags to determine necessity. This eliminates wasteful object instantiation in the common case where no parameters are provided.

2. Reordered error checking: In _handle_onenote_response, the dict error check (isinstance(response, dict) and 'error' in response) is moved before the hasattr(response, 'error') check. This provides a faster path for dict responses, which are more common in API responses.

Why this improves performance:

• Reduced allocations: Line profiler shows the original code always creates RequestConfiguration objects (line 87: 335,484 ns total), while the optimized version only creates them when needed (line 11: 15,976 ns total for 5 hits vs 345 hits).

• Faster error path: The reordered checks in error handling reduce the time spent on hasattr operations for common dict responses.

• Preserved functionality: All error handling, parameter validation, and API behavior remains identical.

Impact on workloads:

The optimization particularly benefits scenarios with:

• High-frequency deletion calls with minimal parameters (most common case)
• Batch operations where many notebooks are deleted with default settings
• Throughput-sensitive applications processing many OneNote operations

The 12% runtime improvement translates directly to faster response times for OneNote deletion operations, especially valuable in enterprise scenarios where OneNote operations may be called frequently for content management or cleanup tasks.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio
from typing import Optional

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

# --- Minimal stubs for dependencies to allow isolated testing ---


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


class DummyNotebookDeleteRequest:
    """
    Simulates the chained call:
    self.client.me.onenote.notebooks.by_notebook_id(notebook_id).delete(request_configuration=config)
    """

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

    async def delete(self, request_configuration=None):
        # delete_behavior is a function or value
        if callable(self.delete_behavior):
            return await self.delete_behavior(self.notebook_id, request_configuration)
        return self.delete_behavior


class DummyNotebooks:
    def __init__(self, delete_behavior):
        self._delete_behavior = delete_behavior

    def by_notebook_id(self, notebook_id):
        return DummyNotebookDeleteRequest(notebook_id, self._delete_behavior)


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


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


class DummyMSGraphServiceClient:
    def __init__(self, delete_behavior):
        self.me = DummyMe(delete_behavior)


class DummyMSGraphClient:
    def __init__(self, delete_behavior):
        self._delete_behavior = delete_behavior

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return DummyMSGraphServiceClient(self._delete_behavior)


# --- TESTS ---

# --- 1. Basic Test Cases ---


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_basic_success():
    """Test basic successful deletion of a notebook."""

    # Simulate a successful delete (returns a dummy object)
    async def success_delete(notebook_id, config):
        # Simulate a successful API response (no error)
        return {"id": notebook_id, "deleted": True}

    client = DummyMSGraphClient(success_delete)
    ds = OneNoteDataSource(client)
    notebook_id = "notebook-123"
    response = await ds.me_onenote_delete_notebooks(notebook_id)


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_basic_with_select_expand():
    """Test deletion with select and expand parameters."""

    async def delete_with_params(notebook_id, config):
        # Check that select/expand are passed correctly
        select = getattr(config.query_parameters, "select", None)
        expand = getattr(config.query_parameters, "expand", None)
        return {"id": notebook_id, "deleted": True, "select": select, "expand": expand}

    client = DummyMSGraphClient(delete_with_params)
    ds = OneNoteDataSource(client)
    notebook_id = "notebook-456"
    response = await ds.me_onenote_delete_notebooks(
        notebook_id, select=["id", "name"], expand=["sections"]
    )


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_basic_with_headers():
    """Test that custom headers are passed and ConsistencyLevel is added for search."""

    async def delete_with_headers(notebook_id, config):
        headers = getattr(config, "headers", {})
        return {}

    client = DummyMSGraphClient(delete_with_headers)
    ds = OneNoteDataSource(client)
    notebook_id = "notebook-789"
    response = await ds.me_onenote_delete_notebooks(
        notebook_id, headers={"My-Header": "Value"}, search="foo"
    )


# --- 2. Edge Test Cases ---


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_notebook_id_empty():
    """Test deletion with empty notebook_id (should still call underlying delete)."""

    async def delete_empty_id(notebook_id, config):
        return {"id": notebook_id, "deleted": True}

    client = DummyMSGraphClient(delete_empty_id)
    ds = OneNoteDataSource(client)
    response = await ds.me_onenote_delete_notebooks("")


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_delete_returns_none():
    """Test when the underlying delete returns None (should be handled as error)."""

    async def delete_none(notebook_id, config):
        return None

    client = DummyMSGraphClient(delete_none)
    ds = OneNoteDataSource(client)
    response = await ds.me_onenote_delete_notebooks("notebook-1")


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_delete_raises_exception():
    """Test when the underlying delete raises an exception."""

    async def delete_raises(notebook_id, config):
        raise RuntimeError("Simulated API failure")

    client = DummyMSGraphClient(delete_raises)
    ds = OneNoteDataSource(client)
    response = await ds.me_onenote_delete_notebooks("notebook-err")


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_response_has_error_attr():
    """Test when the API response has an 'error' attribute (should be handled as error)."""

    class ResponseWithError:
        error = "This is a notebook error"

    async def delete_error(notebook_id, config):
        return ResponseWithError()

    client = DummyMSGraphClient(delete_error)
    ds = OneNoteDataSource(client)
    response = await ds.me_onenote_delete_notebooks("notebook-err2")


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_response_dict_error_key():
    """Test when the API response is a dict with 'error' key."""

    async def delete_error_dict(notebook_id, config):
        return {"error": {"code": "404", "message": "Notebook not found"}}

    client = DummyMSGraphClient(delete_error_dict)
    ds = OneNoteDataSource(client)
    response = await ds.me_onenote_delete_notebooks("notebook-missing")


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_response_has_code_and_message():
    """Test when the API response has 'code' and 'message' attributes."""

    class ResponseWithCodeMessage:
        code = "403"
        message = "Forbidden"

    async def delete_code_message(notebook_id, config):
        return ResponseWithCodeMessage()

    client = DummyMSGraphClient(delete_code_message)
    ds = OneNoteDataSource(client)
    response = await ds.me_onenote_delete_notebooks("notebook-forbidden")


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_concurrent():
    """Test concurrent deletion of multiple notebooks."""

    async def delete_concurrent(notebook_id, config):
        # Return different response per notebook_id
        return {"id": notebook_id, "deleted": True}

    client = DummyMSGraphClient(delete_concurrent)
    ds = OneNoteDataSource(client)
    ids = [f"nb-{i}" for i in range(5)]
    # Run concurrently
    results = await asyncio.gather(
        *[ds.me_onenote_delete_notebooks(nid) for nid in ids]
    )
    for i, r in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_with_all_params():
    """Test passing all possible parameters."""

    async def delete_all_params(notebook_id, config):
        # Check that all params are propagated
        qp = config.query_parameters
        return {"id": notebook_id, "deleted": True}

    client = DummyMSGraphClient(delete_all_params)
    ds = OneNoteDataSource(client)
    response = await ds.me_onenote_delete_notebooks(
        "nb-all",
        select=["id", "name"],
        expand=["sections"],
        filter="name eq 'Test'",
        orderby="name",
        search="Test",
        top=10,
        skip=2,
        headers={"Custom": "yes"},
    )


# --- 3. Large Scale Test Cases ---


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_many_concurrent():
    """Test deleting a large number of notebooks concurrently (scale test, <100)."""

    async def delete_many(notebook_id, config):
        return {"id": notebook_id, "deleted": True}

    client = DummyMSGraphClient(delete_many)
    ds = OneNoteDataSource(client)
    ids = [f"nb-{i}" for i in range(50)]
    results = await asyncio.gather(
        *[ds.me_onenote_delete_notebooks(nid) for nid in ids]
    )
    for i, r in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_large_payloads():
    """Test deleting with large select/expand lists."""

    async def delete_large_lists(notebook_id, config):
        return {"id": notebook_id, "deleted": True}

    client = DummyMSGraphClient(delete_large_lists)
    ds = OneNoteDataSource(client)
    select = [f"prop{i}" for i in range(100)]
    expand = [f"rel{i}" for i in range(100)]
    response = await ds.me_onenote_delete_notebooks(
        "nb-large", select=select, expand=expand
    )


# --- 4. Throughput Test Cases ---


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_throughput_small_load():
    """Throughput: Test small batch of quick deletions."""

    async def delete_quick(notebook_id, config):
        return {"id": notebook_id, "deleted": True}

    client = DummyMSGraphClient(delete_quick)
    ds = OneNoteDataSource(client)
    ids = [f"nb-{i}" for i in range(10)]
    results = await asyncio.gather(
        *[ds.me_onenote_delete_notebooks(nid) for nid in ids]
    )


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_throughput_medium_load():
    """Throughput: Test medium batch of deletions."""

    async def delete_medium(notebook_id, config):
        return {"id": notebook_id, "deleted": True}

    client = DummyMSGraphClient(delete_medium)
    ds = OneNoteDataSource(client)
    ids = [f"nb-{i}" for i in range(50)]
    results = await asyncio.gather(
        *[ds.me_onenote_delete_notebooks(nid) for nid in ids]
    )


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_throughput_high_volume():
    """Throughput: Test high-volume (but bounded) concurrent deletions."""

    async def delete_high(notebook_id, config):
        return {"id": notebook_id, "deleted": True}

    client = DummyMSGraphClient(delete_high)
    ds = OneNoteDataSource(client)
    ids = [f"nb-{i}" for i in range(100)]
    results = await asyncio.gather(
        *[ds.me_onenote_delete_notebooks(nid) for nid 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
# Logger stub

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

# --- Minimal stubs for dependencies ---


class OneNoteResponse:
    """A simple response wrapper for OneNote API responses."""

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


# --- Mock Client Structure ---


class MockNotebookDelete:
    """Simulates the .delete() async method of a notebook resource."""

    def __init__(self, notebook_id, results):
        self.notebook_id = notebook_id
        self.results = results  # Dict of notebook_id -> response

    async def delete(self, request_configuration=None):
        # Simulate async delete operation
        await asyncio.sleep(0)  # Yield control for async realism
        if self.notebook_id in self.results:
            result = self.results[self.notebook_id]
            if isinstance(result, Exception):
                raise result
            return result
        # Simulate not found
        return {"error": {"code": "404", "message": "Notebook not found"}}


class MockNotebooks:
    """Simulates .by_notebook_id() for notebooks."""

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

    def by_notebook_id(self, notebook_id):
        return MockNotebookDelete(notebook_id, self.results)


class MockOneNote:
    """Simulates .notebooks property."""

    def __init__(self, results):
        self.notebooks = MockNotebooks(results)


class MockMe:
    """Simulates .onenote property."""

    def __init__(self, results):
        self.onenote = MockOneNote(results)


class MockMSGraphServiceClient:
    """Simulates the real MS Graph Service Client."""

    def __init__(self, results):
        self.me = MockMe(results)


class MockMSGraphClient:
    """Simulates the MSGraphClient wrapper."""

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

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return MockMSGraphServiceClient(self.results)


# --- TESTS ---

# 1. BASIC TEST CASES


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_success():
    """Test basic successful deletion of a notebook."""
    # Setup: notebook_id "abc" exists and returns a successful response
    results = {"abc": {"id": "abc", "status": "deleted"}}
    ds = OneNoteDataSource(MockMSGraphClient(results))
    response = await ds.me_onenote_delete_notebooks("abc")


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_not_found():
    """Test deletion of a non-existent notebook returns error."""
    results = {}  # No notebooks exist
    ds = OneNoteDataSource(MockMSGraphClient(results))
    response = await ds.me_onenote_delete_notebooks("nonexistent")


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_with_headers_and_query():
    """Test deletion with headers and query parameters."""
    results = {"xyz": {"id": "xyz", "status": "deleted"}}
    ds = OneNoteDataSource(MockMSGraphClient(results))
    headers = {"If-Match": "etag123"}
    response = await ds.me_onenote_delete_notebooks(
        "xyz",
        If_Match="etag123",
        select=["id", "status"],
        expand=["sections"],
        filter="status eq 'active'",
        orderby="id",
        search="Notebook",
        top=1,
        skip=0,
        headers=headers,
    )


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_returns_none():
    """Test when the underlying client returns None (empty response)."""
    results = {"abc": None}
    ds = OneNoteDataSource(MockMSGraphClient(results))
    response = await ds.me_onenote_delete_notebooks("abc")


# 2. EDGE TEST CASES


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_error_object():
    """Test when the client returns an object with an 'error' attribute."""

    class ErrorObj:
        error = "Permission denied"

    results = {"err": ErrorObj()}
    ds = OneNoteDataSource(MockMSGraphClient(results))
    response = await ds.me_onenote_delete_notebooks("err")


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_exception_in_delete():
    """Test when the underlying delete method raises an exception."""
    results = {"fail": RuntimeError("API failure")}
    ds = OneNoteDataSource(MockMSGraphClient(results))
    response = await ds.me_onenote_delete_notebooks("fail")


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_concurrent():
    """Test concurrent deletion of multiple notebooks."""
    results = {
        "a": {"id": "a", "status": "deleted"},
        "b": {"id": "b", "status": "deleted"},
        "c": {"id": "c", "status": "deleted"},
    }
    ds = OneNoteDataSource(MockMSGraphClient(results))
    # Run deletes concurrently
    responses = await asyncio.gather(
        ds.me_onenote_delete_notebooks("a"),
        ds.me_onenote_delete_notebooks("b"),
        ds.me_onenote_delete_notebooks("c"),
    )
    for resp, nid in zip(responses, ["a", "b", "c"]):
        pass


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

    class BadClient:
        def get_client(self):
            return self

        def get_ms_graph_service_client(self):
            return object()

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


# 3. LARGE SCALE TEST CASES


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_many_concurrent():
    """Test deleting many notebooks concurrently (scalability)."""
    notebook_ids = [f"nb_{i}" for i in range(50)]
    results = {nid: {"id": nid, "status": "deleted"} for nid in notebook_ids}
    ds = OneNoteDataSource(MockMSGraphClient(results))
    coros = [ds.me_onenote_delete_notebooks(nid) for nid in notebook_ids]
    responses = await asyncio.gather(*coros)
    for resp, nid in zip(responses, notebook_ids):
        pass


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_mixed_results():
    """Test concurrent deletes with some missing/not found notebooks."""
    notebook_ids = [f"nb_{i}" for i in range(10)]
    # Only even notebooks exist
    results = {
        nid: {"id": nid, "status": "deleted"}
        for nid in notebook_ids
        if int(nid[-1]) % 2 == 0
    }
    ds = OneNoteDataSource(MockMSGraphClient(results))
    coros = [ds.me_onenote_delete_notebooks(nid) for nid in notebook_ids]
    responses = await asyncio.gather(*coros)
    for resp, nid in zip(responses, notebook_ids):
        if int(nid[-1]) % 2 == 0:
            pass
        else:
            pass


# 4. THROUGHPUT TEST CASES


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_throughput_small_load():
    """Throughput: Test small batch of concurrent deletions."""
    notebook_ids = [f"tps_{i}" for i in range(5)]
    results = {nid: {"id": nid, "status": "deleted"} for nid in notebook_ids}
    ds = OneNoteDataSource(MockMSGraphClient(results))
    coros = [ds.me_onenote_delete_notebooks(nid) for nid in notebook_ids]
    responses = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_throughput_medium_load():
    """Throughput: Test medium batch of concurrent deletions."""
    notebook_ids = [f"tpm_{i}" for i in range(25)]
    results = {nid: {"id": nid, "status": "deleted"} for nid in notebook_ids}
    ds = OneNoteDataSource(MockMSGraphClient(results))
    coros = [ds.me_onenote_delete_notebooks(nid) for nid in notebook_ids]
    responses = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_me_onenote_delete_notebooks_throughput_mixed_load():
    """Throughput: Test batch with a mix of successful and failed deletions."""
    notebook_ids = [f"tpmm_{i}" for i in range(20)]
    # Only first 10 exist
    results = {nid: {"id": nid, "status": "deleted"} for nid in notebook_ids[:10]}
    ds = OneNoteDataSource(MockMSGraphClient(results))
    coros = [ds.me_onenote_delete_notebooks(nid) for nid in notebook_ids]
    responses = await asyncio.gather(*coros)
    for i, resp in enumerate(responses):
        if i < 10:
            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_delete_notebooks-mjblhpu5 and push.