⚡️ Speed up method OneNoteDataSource.employee_experience_learning_providers_delete_learning_contents by 11%

[codeflash-ai]

📄 11% (0.11x) speedup for OneNoteDataSource.employee_experience_learning_providers_delete_learning_contents in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 1.56 milliseconds → 1.41 milliseconds (best of 26 runs)

📝 Explanation and details

The optimization replaces expensive object instantiation with lightweight dictionary operations for query parameter handling. The key change is switching from creating a RequestConfiguration() object to build query parameters (which involves multiple attribute assignments) to using a simple dictionary that gets assigned to the config only when needed.

What was optimized:

• Replaced query_params = RequestConfiguration() with query_params = {}
• Changed attribute assignments like query_params.select = ... to dictionary assignments like query_params['select'] = ...
• Added conditional assignment if query_params: before setting config.query_parameters

Why this improves performance:
Object instantiation in Python is expensive due to memory allocation and attribute initialization overhead. The line profiler shows the original RequestConfiguration() call took 750μs (12.3% of total time), while the optimized dictionary creation takes only 189μs (3.4% of total time) - a 75% reduction in this operation alone. Dictionary operations are highly optimized in Python's C implementation, making them significantly faster than object attribute access.

Runtime impact:
The 10% speedup (1.56ms → 1.41ms) comes primarily from eliminating the redundant RequestConfiguration() instantiation and replacing multiple object attribute assignments with efficient dictionary operations. This optimization is particularly effective for the test cases that exercise parameter handling extensively, such as those with multiple query parameters (select, expand, filter, etc.).

The throughput remains unchanged because the optimization affects latency per operation rather than concurrent processing capacity, but the reduced per-call overhead means better resource utilization in high-frequency scenarios.

✅ Correctness verification report:

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

🌀 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 ---


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


# --- Minimal MSGraphClient stub and method chain for testing ---


class DeleteMethodStub:
    def __init__(self, result):
        self._result = result

    async def delete(self, request_configuration=None):
        # Simulate async delete call
        if isinstance(self._result, Exception):
            raise self._result
        return self._result


class LearningContentsStub:
    def __init__(self, result):
        self._result = result

    def by_learningContent_id(self, learningContent_id):
        return DeleteMethodStub(self._result)


class LearningProvidersStub:
    def __init__(self, result):
        self._result = result

    def by_learningProvider_id(self, learningProvider_id):
        return LearningContentsStub(self._result)


class EmployeeExperienceStub:
    def __init__(self, result):
        self._result = result
        self.learning_providers = LearningProvidersStub(self._result)


class ClientStub:
    def __init__(self, result):
        self.employee_experience = EmployeeExperienceStub(result)
        self.me = True  # For hasattr(self.client, "me") check


class MSGraphClientStub:
    def __init__(self, result):
        self._result = result

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return ClientStub(self._result)


# --- UNIT TESTS ---


@pytest.mark.asyncio
async def test_basic_success_response():
    """Test basic async/await behavior and successful delete operation."""
    # Simulate a successful response object (no error attribute)
    response_obj = {"deleted": True, "id": "abc123"}
    ds = OneNoteDataSource(MSGraphClientStub(response_obj))
    result = await ds.employee_experience_learning_providers_delete_learning_contents(
        "provider1", "content1"
    )


@pytest.mark.asyncio
async def test_basic_error_response_with_error_attr():
    """Test when the response has an 'error' attribute (should set success=False)."""

    class Resp:
        error = "Permission denied"

    ds = OneNoteDataSource(MSGraphClientStub(Resp()))
    result = await ds.employee_experience_learning_providers_delete_learning_contents(
        "provider1", "content2"
    )


@pytest.mark.asyncio
async def test_basic_error_response_dict_error():
    """Test when the response is a dict with an 'error' key."""
    response_obj = {"error": {"code": "403", "message": "Forbidden"}}
    ds = OneNoteDataSource(MSGraphClientStub(response_obj))
    result = await ds.employee_experience_learning_providers_delete_learning_contents(
        "provider2", "content3"
    )


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

    class Resp:
        code = "404"
        message = "Not found"

    ds = OneNoteDataSource(MSGraphClientStub(Resp()))
    result = await ds.employee_experience_learning_providers_delete_learning_contents(
        "provider3", "content4"
    )


@pytest.mark.asyncio
async def test_basic_empty_response():
    """Test when the response is None (should return success=False and error message)."""
    ds = OneNoteDataSource(MSGraphClientStub(None))
    result = await ds.employee_experience_learning_providers_delete_learning_contents(
        "provider4", "content5"
    )


@pytest.mark.asyncio
async def test_with_all_parameters():
    """Test passing all optional parameters to ensure they're set correctly."""
    response_obj = {"deleted": True}
    ds = OneNoteDataSource(MSGraphClientStub(response_obj))
    result = await ds.employee_experience_learning_providers_delete_learning_contents(
        "provider5",
        "content6",
        If_Match="etag123",
        select=["id", "name"],
        expand=["sections"],
        filter="name eq 'foo'",
        orderby="name",
        search="notebook",
        top=10,
        skip=2,
        headers={"Custom-Header": "value"},
    )


@pytest.mark.asyncio
async def test_with_select_and_expand_as_str():
    """Test select and expand as single string (should be converted to list)."""
    response_obj = {"deleted": True}
    ds = OneNoteDataSource(MSGraphClientStub(response_obj))
    result = await ds.employee_experience_learning_providers_delete_learning_contents(
        "provider6", "content7", select="id", expand="sections"
    )


@pytest.mark.asyncio
async def test_with_headers_and_search_consistency_level():
    """Test that ConsistencyLevel header is added when search is provided."""
    # The function modifies headers if search is present.
    response_obj = {"deleted": True}
    ds = OneNoteDataSource(MSGraphClientStub(response_obj))
    result = await ds.employee_experience_learning_providers_delete_learning_contents(
        "provider7", "content8", search="test", headers={"X-Test": "1"}
    )


@pytest.mark.asyncio
async def test_exception_in_delete():
    """Test exception is handled if delete raises."""
    error = Exception("Simulated network error")
    ds = OneNoteDataSource(MSGraphClientStub(error))
    result = await ds.employee_experience_learning_providers_delete_learning_contents(
        "provider8", "content9"
    )


@pytest.mark.asyncio
async def test_exception_in_handle_response():
    """Test exception in _handle_onenote_response is handled."""

    # Patch _handle_onenote_response to raise
    class BrokenOneNoteDataSource(OneNoteDataSource):
        def _handle_onenote_response(self, response):
            raise Exception("Broken handler")

    response_obj = {"deleted": True}
    ds = BrokenOneNoteDataSource(MSGraphClientStub(response_obj))
    result = await ds.employee_experience_learning_providers_delete_learning_contents(
        "provider9", "content10"
    )


@pytest.mark.asyncio
async def test_concurrent_deletes():
    """Test concurrent execution of multiple delete operations."""
    response_obj1 = {"deleted": True, "id": "a"}
    response_obj2 = {"deleted": True, "id": "b"}
    ds1 = OneNoteDataSource(MSGraphClientStub(response_obj1))
    ds2 = OneNoteDataSource(MSGraphClientStub(response_obj2))
    # Run two deletes concurrently
    results = await asyncio.gather(
        ds1.employee_experience_learning_providers_delete_learning_contents(
            "providerA", "contentA"
        ),
        ds2.employee_experience_learning_providers_delete_learning_contents(
            "providerB", "contentB"
        ),
    )


@pytest.mark.asyncio
async def test_concurrent_delete_with_error():
    """Test concurrent execution where one delete fails."""
    response_obj = {"deleted": True, "id": "ok"}
    error = Exception("Simulated failure")
    ds1 = OneNoteDataSource(MSGraphClientStub(response_obj))
    ds2 = OneNoteDataSource(MSGraphClientStub(error))
    results = await asyncio.gather(
        ds1.employee_experience_learning_providers_delete_learning_contents(
            "providerC", "contentC"
        ),
        ds2.employee_experience_learning_providers_delete_learning_contents(
            "providerD", "contentD"
        ),
    )


# --- Large Scale Test Cases ---


@pytest.mark.asyncio
async def test_large_scale_multiple_concurrent_deletes():
    """Test many concurrent delete operations for scalability."""
    N = 20
    ds_list = [
        OneNoteDataSource(MSGraphClientStub({"deleted": True, "id": f"id_{i}"}))
        for i in range(N)
    ]
    coros = [
        ds.employee_experience_learning_providers_delete_learning_contents(
            f"provider{i}", f"content{i}"
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_large_scale_some_failures():
    """Test large scale with some failures."""
    N = 10
    ds_list = []
    for i in range(N):
        if i % 3 == 0:
            # Simulate failure
            ds_list.append(OneNoteDataSource(MSGraphClientStub(Exception("fail"))))
        else:
            ds_list.append(
                OneNoteDataSource(MSGraphClientStub({"deleted": True, "id": f"id_{i}"}))
            )
    coros = [
        ds.employee_experience_learning_providers_delete_learning_contents(
            f"provider{i}", f"content{i}"
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*coros)
    for i, r in enumerate(results):
        if i % 3 == 0:
            pass
        else:
            pass


# --- Throughput Test Cases ---


@pytest.mark.asyncio
async def test_OneNoteDataSource_employee_experience_learning_providers_delete_learning_contents_throughput_small_load():
    """Throughput test: small load, should complete quickly."""
    ds = OneNoteDataSource(MSGraphClientStub({"deleted": True}))
    coros = [
        ds.employee_experience_learning_providers_delete_learning_contents(
            "providerX", f"content{i}"
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_OneNoteDataSource_employee_experience_learning_providers_delete_learning_contents_throughput_medium_load():
    """Throughput test: medium load, moderate concurrency."""
    ds = OneNoteDataSource(MSGraphClientStub({"deleted": True}))
    coros = [
        ds.employee_experience_learning_providers_delete_learning_contents(
            "providerY", f"content{i}"
        )
        for i in range(30)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_OneNoteDataSource_employee_experience_learning_providers_delete_learning_contents_throughput_mixed_success_failure():
    """Throughput test: mix of successes and failures."""
    N = 25
    ds_list = []
    for i in range(N):
        if i % 5 == 0:
            ds_list.append(OneNoteDataSource(MSGraphClientStub(Exception("fail"))))
        else:
            ds_list.append(
                OneNoteDataSource(MSGraphClientStub({"deleted": True, "id": f"id_{i}"}))
            )
    coros = [
        ds.employee_experience_learning_providers_delete_learning_contents(
            "providerZ", f"content{i}"
        )
        for i, ds in enumerate(ds_list)
    ]
    results = await asyncio.gather(*coros)
    for i, r in enumerate(results):
        if i % 5 == 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
# Dummy logger for the OneNoteDataSource class
# The function/class under test (EXACT COPY)

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


# Minimal stubs for dependencies (so the tests can run without external libraries)
class OneNoteResponse:
    def __init__(self, success, data=None, error=None):
        self.success = success
        self.data = data
        self.error = error


# Minimal MSGraphClient stub for OneNoteDataSource
class DummyDeleteEndpoint:
    def __init__(self, should_error=False, response=None):
        self.should_error = should_error
        self.response = response

    async def delete(self, request_configuration=None):
        # Simulate error or return the provided response
        if self.should_error:
            raise Exception("Simulated delete error")
        # Simulate a successful delete with a dummy response object
        if self.response is not None:
            return self.response

        # Default success response
        class DummyResponse:
            pass

        return DummyResponse()


class DummyLearningContents:
    def __init__(self, should_error=False, response=None):
        self.should_error = should_error
        self.response = response

    def by_learningContent_id(self, learningContent_id):
        return DummyDeleteEndpoint(self.should_error, self.response)


class DummyLearningProviders:
    def __init__(self, should_error=False, response=None):
        self.should_error = should_error
        self.response = response

    def by_learningProvider_id(self, learningProvider_id):
        return DummyLearningContents(self.should_error, self.response)


class DummyEmployeeExperience:
    def __init__(self, should_error=False, response=None):
        self.should_error = should_error
        self.response = response

        self.learning_providers = DummyLearningProviders(should_error, response)


class DummyClient:
    def __init__(self, should_error=False, response=None):
        self.me = True  # Needed for OneNoteDataSource check
        self.employee_experience = DummyEmployeeExperience(should_error, response)

    def get_ms_graph_service_client(self):
        return self


class DummyMSGraphClient:
    def __init__(self, should_error=False, response=None):
        self.should_error = should_error
        self.response = response

    def get_client(self):
        return DummyClient(self.should_error, self.response)


# ------------------- UNIT TESTS -------------------

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_delete_learning_contents_basic_success():
    """Test basic successful deletion with required parameters."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    result = await datasource.employee_experience_learning_providers_delete_learning_contents(
        learningProvider_id="provider123", learningContent_id="content456"
    )


@pytest.mark.asyncio
async def test_delete_learning_contents_basic_with_optional_params():
    """Test deletion with all optional parameters provided."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    result = await datasource.employee_experience_learning_providers_delete_learning_contents(
        learningProvider_id="provider123",
        learningContent_id="content456",
        If_Match="etagvalue",
        select=["id", "name"],
        expand=["sections"],
        filter="name eq 'test'",
        orderby="name",
        search="searchterm",
        top=10,
        skip=2,
        headers={"Custom-Header": "Value"},
    )


@pytest.mark.asyncio
async def test_delete_learning_contents_basic_with_kwargs():
    """Test deletion with extra kwargs."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    result = await datasource.employee_experience_learning_providers_delete_learning_contents(
        learningProvider_id="provider123",
        learningContent_id="content456",
        irrelevant_param="should_be_ignored",
    )


# 2. Edge Test Cases


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

    class ErrorResponse:
        error = "Some error occurred"

    datasource = OneNoteDataSource(DummyMSGraphClient(response=ErrorResponse()))
    result = await datasource.employee_experience_learning_providers_delete_learning_contents(
        learningProvider_id="provider123", learningContent_id="content456"
    )


@pytest.mark.asyncio
async def test_delete_learning_contents_error_response_dict():
    """Test error handling when response is a dict with error info."""
    error_dict = {"error": {"code": "BadRequest", "message": "Invalid input"}}
    datasource = OneNoteDataSource(DummyMSGraphClient(response=error_dict))
    result = await datasource.employee_experience_learning_providers_delete_learning_contents(
        learningProvider_id="provider123", learningContent_id="content456"
    )


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

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

    datasource = OneNoteDataSource(DummyMSGraphClient(response=CodeMessageResponse()))
    result = await datasource.employee_experience_learning_providers_delete_learning_contents(
        learningProvider_id="provider123", learningContent_id="content456"
    )


@pytest.mark.asyncio
async def test_delete_learning_contents_none_response():
    """Test error handling when response is None."""
    datasource = OneNoteDataSource(DummyMSGraphClient(response=None))
    result = await datasource.employee_experience_learning_providers_delete_learning_contents(
        learningProvider_id="provider123", learningContent_id="content456"
    )


@pytest.mark.asyncio
async def test_delete_learning_contents_exception_in_delete():
    """Test error handling when the delete operation raises an exception."""
    datasource = OneNoteDataSource(DummyMSGraphClient(should_error=True))
    result = await datasource.employee_experience_learning_providers_delete_learning_contents(
        learningProvider_id="provider123", learningContent_id="content456"
    )


@pytest.mark.asyncio
async def test_delete_learning_contents_concurrent_calls():
    """Test concurrent execution of multiple deletions."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    coros = [
        datasource.employee_experience_learning_providers_delete_learning_contents(
            learningProvider_id=f"provider{i}", learningContent_id=f"content{i}"
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_delete_learning_contents_large_scale_concurrent():
    """Test large scale concurrent deletion calls (up to 50)."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    coros = [
        datasource.employee_experience_learning_providers_delete_learning_contents(
            learningProvider_id=f"provider{i}", learningContent_id=f"content{i}"
        )
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_delete_learning_contents_large_scale_error_handling():
    """Test large scale concurrent calls with some errors."""
    # 45 success, 5 error
    datasource_success = OneNoteDataSource(DummyMSGraphClient())
    datasource_error = OneNoteDataSource(DummyMSGraphClient(should_error=True))
    coros = [
        datasource_success.employee_experience_learning_providers_delete_learning_contents(
            learningProvider_id=f"provider{i}", learningContent_id=f"content{i}"
        )
        for i in range(45)
    ] + [
        datasource_error.employee_experience_learning_providers_delete_learning_contents(
            learningProvider_id=f"provider{i}", learningContent_id=f"content{i}"
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)
    success_count = sum(1 for r in results if r.success)
    error_count = sum(1 for r in results if not r.success)


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_delete_learning_contents_throughput_small_load():
    """Throughput test: small load (10 concurrent deletions)."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    coros = [
        datasource.employee_experience_learning_providers_delete_learning_contents(
            learningProvider_id=f"provider{i}", learningContent_id=f"content{i}"
        )
        for i in range(10)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_delete_learning_contents_throughput_medium_load():
    """Throughput test: medium load (50 concurrent deletions)."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    coros = [
        datasource.employee_experience_learning_providers_delete_learning_contents(
            learningProvider_id=f"provider{i}", learningContent_id=f"content{i}"
        )
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_delete_learning_contents_throughput_high_volume():
    """Throughput test: high volume (200 concurrent deletions)."""
    datasource = OneNoteDataSource(DummyMSGraphClient())
    coros = [
        datasource.employee_experience_learning_providers_delete_learning_contents(
            learningProvider_id=f"provider{i}", learningContent_id=f"content{i}"
        )
        for i in range(200)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_delete_learning_contents_throughput_mixed_success_error():
    """Throughput test: mixed success/error (180 success, 20 error)."""
    datasource_success = OneNoteDataSource(DummyMSGraphClient())
    datasource_error = OneNoteDataSource(DummyMSGraphClient(should_error=True))
    coros = [
        datasource_success.employee_experience_learning_providers_delete_learning_contents(
            learningProvider_id=f"provider{i}", learningContent_id=f"content{i}"
        )
        for i in range(180)
    ] + [
        datasource_error.employee_experience_learning_providers_delete_learning_contents(
            learningProvider_id=f"provider{i}", learningContent_id=f"content{i}"
        )
        for i in range(20)
    ]
    results = await asyncio.gather(*coros)
    success_count = sum(1 for r in results if r.success)
    error_count = sum(1 for r in results if not r.success)


# 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.employee_experience_learning_providers_delete_learning_contents-mjh04bp9 and push.