Skip to content

Conversation

@codeflash-ai
Copy link

@codeflash-ai codeflash-ai bot commented Dec 17, 2025

📄 10% (0.10x) speedup for OneNoteDataSource.groups_onenote_notebooks_delete_section_groups in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 1.93 milliseconds 1.76 milliseconds (best of 84 runs)

📝 Explanation and details

The optimized code achieves a 9% runtime speedup through two key micro-optimizations that reduce unnecessary object allocations and redundant assignments:

1. Early Return Pattern in Error Handling (_handle_onenote_response)

  • Original: Set success=True and error_msg=None variables, then conditionally reassign them before creating the response object
  • Optimized: Return OneNoteResponse objects immediately when errors are detected, eliminating redundant variable assignments
  • Impact: Reduces 2 variable assignments per call for error cases and streamlines the happy path execution

2. Conditional Object Creation (groups_onenote_notebooks_delete_section_groups)

  • Original: Always creates RequestConfiguration() objects for query_params and config, even when no parameters are provided
  • Optimized: Only creates these objects when actually needed (when any optional parameters are present)
  • Impact: For calls without optional parameters (which appears common based on test patterns), this eliminates 2 unnecessary object allocations and multiple attribute assignments

Performance Analysis:
The line profiler shows the optimization effects:

  • _handle_onenote_response: Total time reduced from 650μs to 583μs (10% improvement)
  • Main function: Configuration setup time significantly reduced when no parameters are used
  • The 47.6% time spent in the actual API call remains unchanged, which is expected

Test Case Benefits:
Based on the annotated tests, these optimizations are particularly effective for:

  • Basic success cases without optional parameters (most common usage pattern)
  • High-throughput scenarios where the function is called frequently
  • Concurrent operations where reduced object allocation pressure helps overall system performance

The optimizations maintain identical functionality and error handling while reducing computational overhead in the most common execution paths.

Correctness verification report:

Test Status
⚙️ Existing Unit Tests 🔘 None Found
🌀 Generated Regression Tests 517 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

# Dummy logger
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 and response class ---


class OneNoteResponse:
    """Minimal OneNoteResponse stub."""

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


# Simulate the MSGraphClient's nested method chain for delete
class DummySectionGroups:
    def __init__(
        self,
        group_id,
        notebook_id,
        section_group_id,
        delete_result=None,
        should_raise=False,
    ):
        self.group_id = group_id
        self.notebook_id = notebook_id
        self.section_group_id = section_group_id
        self.delete_result = delete_result
        self.should_raise = should_raise

    async def delete(self, request_configuration=None):
        if self.should_raise:
            raise RuntimeError("Simulated API error")
        return self.delete_result


class DummyNotebookId:
    def __init__(self, group_id, notebook_id, delete_result=None, should_raise=False):
        self.group_id = group_id
        self.notebook_id = notebook_id
        self.delete_result = delete_result
        self.should_raise = should_raise

    @property
    def section_groups(self):
        return DummySectionGroupsSelector(
            self.group_id, self.notebook_id, self.delete_result, self.should_raise
        )


class DummySectionGroupsSelector:
    def __init__(self, group_id, notebook_id, delete_result=None, should_raise=False):
        self.group_id = group_id
        self.notebook_id = notebook_id
        self.delete_result = delete_result
        self.should_raise = should_raise

    def by_section_group_id(self, section_group_id):
        return DummySectionGroups(
            self.group_id,
            self.notebook_id,
            section_group_id,
            self.delete_result,
            self.should_raise,
        )


class DummyOnenote:
    def __init__(self, group_id, delete_result=None, should_raise=False):
        self.group_id = group_id
        self.delete_result = delete_result
        self.should_raise = should_raise

    @property
    def notebooks(self):
        return DummyNotebooksSelector(
            self.group_id, self.delete_result, self.should_raise
        )


class DummyNotebooksSelector:
    def __init__(self, group_id, delete_result=None, should_raise=False):
        self.group_id = group_id
        self.delete_result = delete_result
        self.should_raise = should_raise

    def by_notebook_id(self, notebook_id):
        return DummyNotebookId(
            self.group_id, notebook_id, self.delete_result, self.should_raise
        )


class DummyGroups:
    def __init__(self, delete_result=None, should_raise=False):
        self.delete_result = delete_result
        self.should_raise = should_raise

    def by_group_id(self, group_id):
        return DummyGroupId(group_id, self.delete_result, self.should_raise)


class DummyGroupId:
    def __init__(self, group_id, delete_result=None, should_raise=False):
        self.group_id = group_id
        self.delete_result = delete_result
        self.should_raise = should_raise

    @property
    def onenote(self):
        return DummyOnenote(self.group_id, self.delete_result, self.should_raise)


class DummyClient:
    """Simulates the full client chain."""

    def __init__(self, delete_result=None, should_raise=False):
        self._delete_result = delete_result
        self._should_raise = should_raise
        self.groups = DummyGroups(delete_result, should_raise)
        self.me = True  # for hasattr(self.client, "me")

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return self


# --- Unit Tests ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_delete_section_groups_basic_success():
    """Test basic successful deletion with valid IDs."""
    # Simulate a successful API delete response (could be anything, e.g., None or an object)
    dummy_response = object()
    client = DummyClient(delete_result=dummy_response)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_onenote_notebooks_delete_section_groups("gid", "nid", "sgid")


@pytest.mark.asyncio
async def test_delete_section_groups_returns_error_on_none_response():
    """Test that a None response from API is handled as error."""
    client = DummyClient(delete_result=None)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_onenote_notebooks_delete_section_groups("gid", "nid", "sgid")


@pytest.mark.asyncio
async def test_delete_section_groups_returns_error_field_in_dict():
    """Test that a dict with 'error' key is handled as error."""
    error_dict = {"error": {"code": "BadRequest", "message": "Invalid section group"}}
    client = DummyClient(delete_result=error_dict)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_onenote_notebooks_delete_section_groups("gid", "nid", "sgid")


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

    class DummyErrorObj:
        error = "Some error string"

    client = DummyClient(delete_result=DummyErrorObj())
    ds = OneNoteDataSource(client)
    resp = await ds.groups_onenote_notebooks_delete_section_groups("gid", "nid", "sgid")


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

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

    client = DummyClient(delete_result=DummyCodeMsgObj())
    ds = OneNoteDataSource(client)
    resp = await ds.groups_onenote_notebooks_delete_section_groups("gid", "nid", "sgid")


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_delete_section_groups_concurrent_calls():
    """Test concurrent deletion calls with different IDs."""
    dummy_resp1 = object()
    dummy_resp2 = object()
    dummy_resp3 = object()
    # Each client instance returns a different response
    ds1 = OneNoteDataSource(DummyClient(delete_result=dummy_resp1))
    ds2 = OneNoteDataSource(DummyClient(delete_result=dummy_resp2))
    ds3 = OneNoteDataSource(DummyClient(delete_result=dummy_resp3))
    # Run concurrently
    results = await asyncio.gather(
        ds1.groups_onenote_notebooks_delete_section_groups("g1", "n1", "sg1"),
        ds2.groups_onenote_notebooks_delete_section_groups("g2", "n2", "sg2"),
        ds3.groups_onenote_notebooks_delete_section_groups("g3", "n3", "sg3"),
    )


@pytest.mark.asyncio
async def test_delete_section_groups_exception_handling():
    """Test that exceptions in the API call are wrapped as error responses."""
    # Simulate the delete method raising an exception
    client = DummyClient(should_raise=True)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_onenote_notebooks_delete_section_groups("gid", "nid", "sgid")


@pytest.mark.asyncio
async def test_delete_section_groups_with_all_optional_params():
    """Test providing all optional parameters."""
    dummy_response = object()
    client = DummyClient(delete_result=dummy_response)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_onenote_notebooks_delete_section_groups(
        "gid",
        "nid",
        "sgid",
        If_Match="etag",
        select=["id", "name"],
        expand=["sections"],
        filter="name eq 'Test'",
        orderby="name",
        search="notebook",
        top=5,
        skip=2,
        headers={"Custom-Header": "Value"},
    )


@pytest.mark.asyncio
async def test_delete_section_groups_with_kwargs():
    """Test that extra kwargs do not break the function."""
    dummy_response = object()
    client = DummyClient(delete_result=dummy_response)
    ds = OneNoteDataSource(client)
    resp = await ds.groups_onenote_notebooks_delete_section_groups(
        "gid", "nid", "sgid", custom_param="custom"
    )


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_delete_section_groups_many_concurrent():
    """Test many concurrent deletion calls for scalability."""
    dummy_responses = [object() for _ in range(20)]
    data_sources = [
        OneNoteDataSource(DummyClient(delete_result=resp)) for resp in dummy_responses
    ]
    tasks = [
        ds.groups_onenote_notebooks_delete_section_groups(f"g{i}", f"n{i}", f"sg{i}")
        for i, ds in enumerate(data_sources)
    ]
    results = await asyncio.gather(*tasks)
    # Ensure each response matches its dummy object
    for i, r in enumerate(results):
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_notebooks_delete_section_groups_throughput_small_load():
    """Throughput: Run a small batch of deletions and ensure all succeed quickly."""
    dummy_responses = [object() for _ in range(5)]
    data_sources = [
        OneNoteDataSource(DummyClient(delete_result=resp)) for resp in dummy_responses
    ]
    tasks = [
        ds.groups_onenote_notebooks_delete_section_groups(f"g{i}", f"n{i}", f"sg{i}")
        for i, ds in enumerate(data_sources)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_notebooks_delete_section_groups_throughput_medium_load():
    """Throughput: Run a medium batch of deletions and ensure all succeed."""
    dummy_responses = [object() for _ in range(30)]
    data_sources = [
        OneNoteDataSource(DummyClient(delete_result=resp)) for resp in dummy_responses
    ]
    tasks = [
        ds.groups_onenote_notebooks_delete_section_groups(f"g{i}", f"n{i}", f"sg{i}")
        for i, ds in enumerate(data_sources)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_notebooks_delete_section_groups_throughput_high_volume():
    """Throughput: Run a high volume of deletions (but <1000) and ensure all succeed."""
    count = 100
    dummy_responses = [object() for _ in range(count)]
    data_sources = [
        OneNoteDataSource(DummyClient(delete_result=resp)) for resp in dummy_responses
    ]
    tasks = [
        ds.groups_onenote_notebooks_delete_section_groups(f"g{i}", f"n{i}", f"sg{i}")
        for i, ds in enumerate(data_sources)
    ]
    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.
import asyncio
from typing import Optional

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

# --- Minimal stubs for dependencies ---


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


class DummyDelete:
    """Simulates the .delete(request_configuration=...) async method."""

    def __init__(self, behavior="success"):
        self.behavior = behavior

    async def delete(self, request_configuration=None):
        # Simulate different behaviors for test coverage
        if self.behavior == "success":
            return {"deleted": True}
        elif self.behavior == "error":

            class ErrObj:
                error = "Simulated error"

            return ErrObj()
        elif self.behavior == "dict_error":
            return {"error": {"code": "404", "message": "Not found"}}
        elif self.behavior == "exception":
            raise RuntimeError("Simulated exception")
        elif self.behavior == "none":
            return None
        elif self.behavior == "custom_response":

            class Resp:
                code = 400
                message = "Bad request"

            return Resp()
        else:
            return {"deleted": True}


class DummySectionGroups:
    def __init__(self, behavior="success"):
        self.behavior = behavior

    def by_section_group_id(self, sectionGroup_id):
        return DummyDelete(self.behavior)


class DummyNotebooks:
    def __init__(self, behavior="success"):
        self.behavior = behavior

    def by_notebook_id(self, notebook_id):
        return DummySectionGroups(self.behavior)


class DummyOnenote:
    def __init__(self, behavior="success"):
        self.behavior = behavior

    @property
    def notebooks(self):
        return DummyNotebooks(self.behavior)


class DummyGroups:
    def __init__(self, behavior="success"):
        self.behavior = behavior

    def by_group_id(self, group_id):
        class Dummy:
            def __init__(self, behavior):
                self.behavior = behavior

            @property
            def onenote(self):
                return DummyOnenote(self.behavior)

        return Dummy(self.behavior)


class DummyClient:
    def __init__(self, behavior="success"):
        self.behavior = behavior
        self.me = True  # to pass hasattr check
        self.groups = DummyGroups(behavior=self.behavior)

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return self


# --- TESTS ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_delete_section_groups_success():
    """Test basic successful deletion returns OneNoteResponse with success=True."""
    ds = OneNoteDataSource(DummyClient(behavior="success"))
    resp = await ds.groups_onenote_notebooks_delete_section_groups(
        group_id="group1", notebook_id="nb1", sectionGroup_id="sg1"
    )


@pytest.mark.asyncio
async def test_delete_section_groups_with_all_parameters():
    """Test passing all optional parameters and custom headers."""
    ds = OneNoteDataSource(DummyClient(behavior="success"))
    resp = await ds.groups_onenote_notebooks_delete_section_groups(
        group_id="group1",
        notebook_id="nb1",
        sectionGroup_id="sg1",
        If_Match="etag123",
        select=["id", "name"],
        expand=["sections"],
        filter="name eq 'Test'",
        orderby="name",
        search="notes",
        top=5,
        skip=2,
        headers={"Authorization": "Bearer token"},
    )


@pytest.mark.asyncio
async def test_delete_section_groups_returns_error_object():
    """Test response object with .error attribute is handled as error."""
    ds = OneNoteDataSource(DummyClient(behavior="error"))
    resp = await ds.groups_onenote_notebooks_delete_section_groups(
        group_id="group1", notebook_id="nb1", sectionGroup_id="sg1"
    )


@pytest.mark.asyncio
async def test_delete_section_groups_returns_dict_error():
    """Test response as dict with 'error' key is handled as error."""
    ds = OneNoteDataSource(DummyClient(behavior="dict_error"))
    resp = await ds.groups_onenote_notebooks_delete_section_groups(
        group_id="group1", notebook_id="nb1", sectionGroup_id="sg1"
    )


@pytest.mark.asyncio
async def test_delete_section_groups_returns_none():
    """Test None response returns error with correct message."""
    ds = OneNoteDataSource(DummyClient(behavior="none"))
    resp = await ds.groups_onenote_notebooks_delete_section_groups(
        group_id="group1", notebook_id="nb1", sectionGroup_id="sg1"
    )


@pytest.mark.asyncio
async def test_delete_section_groups_returns_custom_response():
    """Test response with .code and .message attributes is handled as error."""
    ds = OneNoteDataSource(DummyClient(behavior="custom_response"))
    resp = await ds.groups_onenote_notebooks_delete_section_groups(
        group_id="group1", notebook_id="nb1", sectionGroup_id="sg1"
    )


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_delete_section_groups_exception_in_delete():
    """Test exception in delete call is caught and wrapped in OneNoteResponse."""
    ds = OneNoteDataSource(DummyClient(behavior="exception"))
    resp = await ds.groups_onenote_notebooks_delete_section_groups(
        group_id="group1", notebook_id="nb1", sectionGroup_id="sg1"
    )


@pytest.mark.asyncio
async def test_delete_section_groups_invalid_client():
    """Test ValueError is raised if client does not have 'me' attribute."""

    class NoMeClient:
        def get_client(self):
            return self

        def get_ms_graph_service_client(self):
            return self

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


@pytest.mark.asyncio
async def test_delete_section_groups_concurrent_calls():
    """Test multiple concurrent deletions with different behaviors."""
    ds_success = OneNoteDataSource(DummyClient(behavior="success"))
    ds_error = OneNoteDataSource(DummyClient(behavior="error"))
    tasks = [
        ds_success.groups_onenote_notebooks_delete_section_groups(
            group_id="g1", notebook_id="n1", sectionGroup_id="s1"
        ),
        ds_error.groups_onenote_notebooks_delete_section_groups(
            group_id="g2", notebook_id="n2", sectionGroup_id="s2"
        ),
    ]
    results = await asyncio.gather(*tasks)


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_delete_section_groups_many_concurrent():
    """Test 50 concurrent successful deletions."""
    ds = OneNoteDataSource(DummyClient(behavior="success"))
    tasks = [
        ds.groups_onenote_notebooks_delete_section_groups(
            group_id=f"g{i}", notebook_id=f"n{i}", sectionGroup_id=f"s{i}"
        )
        for i in range(50)
    ]
    results = await asyncio.gather(*tasks)
    for r in results:
        pass


@pytest.mark.asyncio
async def test_delete_section_groups_many_concurrent_mixed():
    """Test 20 concurrent deletions, half success, half error."""
    ds_success = OneNoteDataSource(DummyClient(behavior="success"))
    ds_error = OneNoteDataSource(DummyClient(behavior="error"))
    tasks = []
    for i in range(10):
        tasks.append(
            ds_success.groups_onenote_notebooks_delete_section_groups(
                group_id=f"g{i}", notebook_id=f"n{i}", sectionGroup_id=f"s{i}"
            )
        )
        tasks.append(
            ds_error.groups_onenote_notebooks_delete_section_groups(
                group_id=f"g{i + 10}",
                notebook_id=f"n{i + 10}",
                sectionGroup_id=f"s{i + 10}",
            )
        )
    results = await asyncio.gather(*tasks)
    for i, r in enumerate(results):
        if i % 2 == 0:
            pass
        else:
            pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_notebooks_delete_section_groups_throughput_small_load():
    """Throughput test: 10 sequential deletions."""
    ds = OneNoteDataSource(DummyClient(behavior="success"))
    for i in range(10):
        resp = await ds.groups_onenote_notebooks_delete_section_groups(
            group_id=f"g{i}", notebook_id=f"n{i}", sectionGroup_id=f"s{i}"
        )


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_notebooks_delete_section_groups_throughput_medium_load():
    """Throughput test: 50 concurrent deletions."""
    ds = OneNoteDataSource(DummyClient(behavior="success"))
    tasks = [
        ds.groups_onenote_notebooks_delete_section_groups(
            group_id=f"g{i}", notebook_id=f"n{i}", sectionGroup_id=f"s{i}"
        )
        for i in range(50)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_OneNoteDataSource_groups_onenote_notebooks_delete_section_groups_throughput_high_error_rate():
    """Throughput test: 30 concurrent deletions, all fail."""
    ds = OneNoteDataSource(DummyClient(behavior="error"))
    tasks = [
        ds.groups_onenote_notebooks_delete_section_groups(
            group_id=f"g{i}", notebook_id=f"n{i}", sectionGroup_id=f"s{i}"
        )
        for i in range(30)
    ]
    results = await asyncio.gather(*tasks)
    for r in results:
        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.groups_onenote_notebooks_delete_section_groups-mjaju53o and push.

Codeflash Static Badge

…oups

The optimized code achieves a **9% runtime speedup** through two key micro-optimizations that reduce unnecessary object allocations and redundant assignments:

**1. Early Return Pattern in Error Handling (`_handle_onenote_response`)**
- **Original**: Set `success=True` and `error_msg=None` variables, then conditionally reassign them before creating the response object
- **Optimized**: Return `OneNoteResponse` objects immediately when errors are detected, eliminating redundant variable assignments
- **Impact**: Reduces 2 variable assignments per call for error cases and streamlines the happy path execution

**2. Conditional Object Creation (`groups_onenote_notebooks_delete_section_groups`)**
- **Original**: Always creates `RequestConfiguration()` objects for `query_params` and `config`, even when no parameters are provided
- **Optimized**: Only creates these objects when actually needed (when any optional parameters are present)
- **Impact**: For calls without optional parameters (which appears common based on test patterns), this eliminates 2 unnecessary object allocations and multiple attribute assignments

**Performance Analysis:**
The line profiler shows the optimization effects:
- `_handle_onenote_response`: Total time reduced from 650μs to 583μs (10% improvement)
- Main function: Configuration setup time significantly reduced when no parameters are used
- The 47.6% time spent in the actual API call remains unchanged, which is expected

**Test Case Benefits:**
Based on the annotated tests, these optimizations are particularly effective for:
- **Basic success cases** without optional parameters (most common usage pattern)
- **High-throughput scenarios** where the function is called frequently
- **Concurrent operations** where reduced object allocation pressure helps overall system performance

The optimizations maintain identical functionality and error handling while reducing computational overhead in the most common execution paths.
@codeflash-ai codeflash-ai bot requested a review from mashraf-222 December 17, 2025 21:54
@codeflash-ai codeflash-ai bot added ⚡️ codeflash Optimization PR opened by Codeflash AI 🎯 Quality: High Optimization Quality according to Codeflash labels Dec 17, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

⚡️ codeflash Optimization PR opened by Codeflash AI 🎯 Quality: High Optimization Quality according to Codeflash

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant