⚡️ Speed up method OneNoteDataSource.users_delete_onenote by 7%

[codeflash-ai]

📄 7% (0.07x) speedup for OneNoteDataSource.users_delete_onenote in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 894 microseconds → 833 microseconds (best of 164 runs)

📝 Explanation and details

The optimized code achieves a 7% runtime improvement through two key optimizations:

1. Streamlined Error Handling in _handle_onenote_response
The original code used intermediate variables (success = True, error_msg = None) and a single return statement at the end. The optimized version eliminates these variables by returning immediately when error conditions are detected. This reduces:

• Variable assignments (39.1% → 47.7% time spent in return statements shows more direct execution)
• Conditional logic overhead
• Memory allocations for temporary variables

2. Conditional Query Parameter Creation in users_delete_onenote
The original code always created a RequestConfiguration() object for query parameters, even when no parameters were provided. The optimized version only creates this object when actual query parameters exist:

# Original: Always creates object
query_params = RequestConfiguration()

# Optimized: Only creates when needed
query_params = None
if select or expand or filter or orderby or search or (top is not None) or (skip is not None):
    query_params = RequestConfiguration()

This optimization is particularly effective for the common case where most API calls don't use query parameters, avoiding unnecessary object instantiation overhead (10.3% → 1.6% time reduction in query parameter handling).

Performance Impact Analysis:
The line profiler shows the optimizations work best when:

• Most responses are successful (early returns avoid variable assignments)
• API calls use minimal query parameters (avoiding unnecessary object creation)

The test results demonstrate consistent improvements across basic success cases, error handling scenarios, and concurrent operations. The optimization maintains identical functionality while reducing computational overhead in the most common execution paths.

Note on Throughput: While runtime improved 7%, throughput shows a slight decrease (-1.8%). This apparent contradiction likely reflects measurement variance in async operations, where the timing of individual calls versus aggregate throughput can differ due to concurrency effects and test environment factors.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 458 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
from typing import Optional

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

# --- Minimal stubs for dependencies ---


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

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


class DummyDelete:
    """Dummy async delete method to simulate OneNote API."""

    def __init__(self, user_id, behavior=None):
        self.user_id = user_id
        self.behavior = behavior

    async def delete(self, request_configuration=None):
        # Simulate different behaviors based on user_id and test scenario
        if self.behavior == "error":
            raise Exception("Simulated API error")
        if self.behavior == "none":
            return None
        if self.behavior == "dict_error":
            return {"error": {"code": "404", "message": "Not Found"}}
        if self.behavior == "attr_error":

            class Resp:
                pass

            resp = Resp()
            resp.error = "Attr error"
            return resp
        if self.behavior == "code_message":

            class Resp:
                pass

            resp = Resp()
            resp.code = "403"
            resp.message = "Forbidden"
            return resp
        # Simulate normal success response
        return {"deleted": True, "user_id": self.user_id}


class DummyByUserId:
    """Dummy by_user_id for simulating user lookup."""

    def __init__(self, behavior=None):
        self.behavior = behavior

    def onenote(self):
        return DummyDelete(user_id="dummy", behavior=self.behavior)

    # For compatibility with .by_user_id().onenote.delete
    def by_user_id(self, user_id):
        return DummyOnenote(user_id, self.behavior)


class DummyOnenote:
    """Dummy onenote object for simulating delete."""

    def __init__(self, user_id, behavior=None):
        self.user_id = user_id
        self.behavior = behavior

    async def delete(self, request_configuration=None):
        # Use DummyDelete logic
        return await DummyDelete(self.user_id, self.behavior).delete(
            request_configuration
        )


class DummyClient:
    """Dummy client to simulate the GraphServiceClient."""

    def __init__(self, behavior=None):
        self.behavior = behavior
        self.me = True  # Required attribute for OneNoteDataSource

    def users(self):
        return DummyByUserId(self.behavior)

    def by_user_id(self, user_id):
        return DummyOnenote(user_id, self.behavior)

    # For .onenote.delete
    def onenote(self):
        return DummyDelete("dummy", self.behavior)


class DummyMSGraphClient:
    """Dummy MSGraphClient for OneNoteDataSource."""

    def __init__(self, behavior=None):
        self.behavior = behavior

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return DummyClient(self.behavior)


# --- Unit tests ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_users_delete_onenote_basic_success():
    """Basic test: Successful deletion returns success response."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    resp = await ds.users_delete_onenote("user123")


@pytest.mark.asyncio
async def test_users_delete_onenote_basic_with_select_expand():
    """Basic test: select and expand parameters are handled."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    resp = await ds.users_delete_onenote(
        "user456", select=["notebook"], expand=["sections"]
    )


@pytest.mark.asyncio
async def test_users_delete_onenote_basic_with_headers():
    """Basic test: custom headers are passed through."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    headers = {"Authorization": "Bearer token"}
    resp = await ds.users_delete_onenote("user789", headers=headers)


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_users_delete_onenote_none_response():
    """Edge: API returns None, should be handled as error."""
    client = DummyMSGraphClient(behavior="none")
    ds = OneNoteDataSource(client)
    resp = await ds.users_delete_onenote("user_none")


@pytest.mark.asyncio
async def test_users_delete_onenote_dict_error_response():
    """Edge: API returns dict with error key, handled as error."""
    client = DummyMSGraphClient(behavior="dict_error")
    ds = OneNoteDataSource(client)
    resp = await ds.users_delete_onenote("user_dict_error")


@pytest.mark.asyncio
async def test_users_delete_onenote_attr_error_response():
    """Edge: API returns object with error attribute, handled as error."""
    client = DummyMSGraphClient(behavior="attr_error")
    ds = OneNoteDataSource(client)
    resp = await ds.users_delete_onenote("user_attr_error")


@pytest.mark.asyncio
async def test_users_delete_onenote_code_message_response():
    """Edge: API returns object with code/message attributes, handled as error."""
    client = DummyMSGraphClient(behavior="code_message")
    ds = OneNoteDataSource(client)
    resp = await ds.users_delete_onenote("user_code_message")


@pytest.mark.asyncio
async def test_users_delete_onenote_exception_handling():
    """Edge: API raises exception, should be handled gracefully."""
    client = DummyMSGraphClient(behavior="error")
    ds = OneNoteDataSource(client)
    resp = await ds.users_delete_onenote("user_error")


@pytest.mark.asyncio
async def test_users_delete_onenote_concurrent_execution():
    """Edge: Multiple concurrent deletions should all succeed."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    user_ids = [f"user_{i}" for i in range(5)]
    coros = [ds.users_delete_onenote(uid) for uid in user_ids]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_users_delete_onenote_invalid_client_attribute():
    """Edge: client missing 'me' attribute should raise ValueError."""

    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_users_delete_onenote_large_scale_concurrent():
    """Large scale: 50 concurrent deletions are handled correctly."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    user_ids = [f"user_{i}" for i in range(50)]
    coros = [ds.users_delete_onenote(uid) for uid in user_ids]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_users_delete_onenote_large_scale_with_varied_args():
    """Large scale: 20 concurrent deletions with varied args."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    user_ids = [f"user_{i}" for i in range(20)]
    coros = [
        ds.users_delete_onenote(
            uid,
            select=["notebook"] if i % 2 == 0 else None,
            expand=["sections"] if i % 3 == 0 else None,
            top=i,
            skip=0,
        )
        for i, uid in enumerate(user_ids)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_users_delete_onenote_throughput_small_load():
    """Throughput: Small load of 10 concurrent deletions."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    user_ids = [f"user_{i}" for i in range(10)]
    coros = [ds.users_delete_onenote(uid) for uid in user_ids]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_users_delete_onenote_throughput_medium_load():
    """Throughput: Medium load of 100 concurrent deletions."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    user_ids = [f"user_{i}" for i in range(100)]
    coros = [ds.users_delete_onenote(uid) for uid in user_ids]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_users_delete_onenote_throughput_with_search_and_headers():
    """Throughput: 30 concurrent deletions with search and custom headers."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    user_ids = [f"user_{i}" for i in range(30)]
    coros = [
        ds.users_delete_onenote(uid, search="notebook", headers={"X-Test": "true"})
        for uid in user_ids
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_users_delete_onenote_throughput_error_handling():
    """Throughput: 10 concurrent deletions, half with simulated errors."""
    client_ok = DummyMSGraphClient()
    client_err = DummyMSGraphClient(behavior="error")
    ds_ok = OneNoteDataSource(client_ok)
    ds_err = OneNoteDataSource(client_err)
    coros = []
    for i in range(5):
        coros.append(ds_ok.users_delete_onenote(f"user_ok_{i}"))
        coros.append(ds_err.users_delete_onenote(f"user_err_{i}"))
    results = await asyncio.gather(*coros)
    for i in range(10):
        resp = results[i]
        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
# Dummy logger to avoid errors in _handle_onenote_response
# --- Copy of the function under test (as required, UNMODIFIED) ---

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


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


# Dummy client to simulate the MSGraphClient
class DummyDeleteCall:
    def __init__(self, response=None, raise_exc=None):
        self._response = response
        self._raise_exc = raise_exc

    async def delete(self, request_configuration=None):
        if self._raise_exc:
            raise self._raise_exc
        return self._response


class DummyByUserId:
    def __init__(self, delete_call):
        self._delete_call = delete_call

    @property
    def onenote(self):
        return self._delete_call


class DummyUsers:
    def __init__(self, delete_call):
        self._delete_call = delete_call

    def by_user_id(self, user_id):
        return DummyByUserId(self._delete_call)


class DummyClient:
    def __init__(self, delete_call):
        self.me = True  # to pass hasattr check
        self.users = DummyUsers(delete_call)


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

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return self._client


# --- UNIT TESTS ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_users_delete_onenote_basic_success():
    """Test basic successful delete operation returns OneNoteResponse(success=True)."""

    # Simulate a successful delete response (could be None or a dummy object)
    class DummyResponse:
        pass

    dummy_response = DummyResponse()
    delete_call = DummyDeleteCall(response=dummy_response)
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    result = await ds.users_delete_onenote("user123")


@pytest.mark.asyncio
async def test_users_delete_onenote_basic_error_response_object():
    """Test error handling when response has an 'error' attribute."""

    class DummyErrorResponse:
        error = "Something went wrong"

    delete_call = DummyDeleteCall(response=DummyErrorResponse())
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    result = await ds.users_delete_onenote("user456")


@pytest.mark.asyncio
async def test_users_delete_onenote_basic_error_dict():
    """Test error handling when response is a dict with 'error' key."""
    error_dict = {"error": {"code": "400", "message": "Bad request"}}
    delete_call = DummyDeleteCall(response=error_dict)
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    result = await ds.users_delete_onenote("user789")


@pytest.mark.asyncio
async def test_users_delete_onenote_basic_error_code_message():
    """Test error handling when response has 'code' and 'message' attributes."""

    class DummyCodeMessage:
        code = "401"
        message = "Unauthorized"

    delete_call = DummyDeleteCall(response=DummyCodeMessage())
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    result = await ds.users_delete_onenote("user000")


@pytest.mark.asyncio
async def test_users_delete_onenote_basic_none_response():
    """Test handling when the delete returns None (should be treated as error)."""
    delete_call = DummyDeleteCall(response=None)
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    result = await ds.users_delete_onenote("user111")


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_users_delete_onenote_exception_handling():
    """Test that exceptions in delete are caught and returned as error in OneNoteResponse."""

    class CustomException(Exception):
        pass

    delete_call = DummyDeleteCall(raise_exc=CustomException("Simulated failure"))
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    result = await ds.users_delete_onenote("user222")


@pytest.mark.asyncio
async def test_users_delete_onenote_with_all_parameters():
    """Test passing all parameters including headers, select, expand, filter, orderby, search, top, skip."""

    class DummyResponse:
        pass

    delete_call = DummyDeleteCall(response=DummyResponse())
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    result = await ds.users_delete_onenote(
        user_id="user333",
        If_Match="etag",
        select=["id", "name"],
        expand=["sections"],
        filter="name eq 'Test'",
        orderby="name desc",
        search="notebook",
        top=5,
        skip=1,
        headers={"Custom-Header": "value"},
    )


@pytest.mark.asyncio
async def test_users_delete_onenote_select_and_expand_as_str():
    """Test select/expand parameters passed as a single string instead of list."""

    class DummyResponse:
        pass

    delete_call = DummyDeleteCall(response=DummyResponse())
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    result = await ds.users_delete_onenote(
        user_id="user444", select="id", expand="sections"
    )


@pytest.mark.asyncio
async def test_users_delete_onenote_headers_none_and_search():
    """Test search parameter with headers=None to ensure ConsistencyLevel is set."""

    class DummyResponse:
        pass

    captured_headers = {}

    class CustomDeleteCall(DummyDeleteCall):
        async def delete(self, request_configuration=None):
            # Capture headers for assertion
            if (
                hasattr(request_configuration, "headers")
                and request_configuration.headers
            ):
                captured_headers.update(request_configuration.headers)
            return DummyResponse()

    delete_call = CustomDeleteCall(response=DummyResponse())
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    await ds.users_delete_onenote(user_id="user555", search="findme", headers=None)


@pytest.mark.asyncio
async def test_users_delete_onenote_concurrent():
    """Test concurrent execution of users_delete_onenote to ensure async safety."""

    class DummyResponse:
        pass

    delete_call = DummyDeleteCall(response=DummyResponse())
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    # Run 10 concurrent deletions
    results = await asyncio.gather(
        *[ds.users_delete_onenote(f"user-{i}") for i in range(10)]
    )
    for result in results:
        pass


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_users_delete_onenote_large_scale_concurrent():
    """Test large scale concurrent execution (e.g., 50 users)."""

    class DummyResponse:
        pass

    delete_call = DummyDeleteCall(response=DummyResponse())
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    # 50 concurrent calls (well under 1000)
    results = await asyncio.gather(
        *[ds.users_delete_onenote(f"user-{i}") for i in range(50)]
    )
    for result in results:
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_users_delete_onenote_throughput_small_load():
    """Throughput test: small load (5 concurrent deletions)."""

    class DummyResponse:
        pass

    delete_call = DummyDeleteCall(response=DummyResponse())
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    tasks = [ds.users_delete_onenote(f"user-tp-{i}") for i in range(5)]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_users_delete_onenote_throughput_medium_load():
    """Throughput test: medium load (20 concurrent deletions)."""

    class DummyResponse:
        pass

    delete_call = DummyDeleteCall(response=DummyResponse())
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    tasks = [ds.users_delete_onenote(f"user-tp-{i}") for i in range(20)]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_users_delete_onenote_throughput_large_load():
    """Throughput test: large load (100 concurrent deletions)."""

    class DummyResponse:
        pass

    delete_call = DummyDeleteCall(response=DummyResponse())
    client = DummyMSGraphClient(delete_call)
    ds = OneNoteDataSource(client)
    tasks = [ds.users_delete_onenote(f"user-tp-{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.

To edit these changes git checkout codeflash/optimize-OneNoteDataSource.users_delete_onenote-mjaerqia and push.