Skip to content

Conversation

@codeflash-ai
Copy link

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

📄 12% (0.12x) speedup for OneNoteDataSource.me_onenote_section_groups_delete_sections in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 2.38 milliseconds 2.13 milliseconds (best of 76 runs)

📝 Explanation and details

The optimized code achieves an 11% runtime improvement and 1.3% throughput increase by reducing object instantiation overhead in the query parameter handling logic.

Key optimization applied:

  • Replaced double RequestConfiguration() instantiation with a single instance - The original code created one RequestConfiguration for query_params and another for config, then always assigned the first to the second regardless of whether any parameters were set.
  • Used a lightweight dictionary for parameter collection - Instead of immediately creating a RequestConfiguration object and setting its attributes, parameters are first collected in a plain dictionary (query_params = {}).
  • Conditional assignment of query parameters - The config.query_parameters assignment only happens when there are actual parameters to set (if query_params:), avoiding unnecessary object attribute assignments.

Why this leads to speedup:

  • Reduced object creation overhead - RequestConfiguration() instantiation involves class initialization overhead that's eliminated for the first object
  • Fewer attribute assignments - Dictionary operations (query_params["key"] = value) are faster than object attribute assignments (query_params.key = value)
  • Conditional processing - The optimization only assigns query_parameters when needed, avoiding unnecessary work when no query parameters are present

Performance characteristics:

  • The line profiler shows the original RequestConfiguration() call took 1.11ms vs 0.26ms for the dictionary initialization
  • Most effective for calls with few or no query parameters (which appears to be the majority based on test patterns)
  • The async nature means this optimization compounds across concurrent operations, explaining the throughput improvement

This optimization is particularly valuable for high-frequency OneNote API operations where query parameters are often sparse or absent.

Correctness verification report:

Test Status
⚙️ Existing Unit Tests 🔘 None Found
🌀 Generated Regression Tests 1208 Passed
⏪ Replay Tests 🔘 None Found
🔎 Concolic Coverage Tests 🔘 None Found
📊 Tests Coverage 92.9%
🌀 Generated Regression Tests and Runtime
import asyncio

# Dummy logger
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:
    """Dummy async delete method to simulate MS Graph SDK call."""

    def __init__(self, response=None, raise_exc: Exception = 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 DummySections:
    def __init__(self, response=None, raise_exc: Exception = None):
        self._response = response
        self._raise_exc = raise_exc

    def by_onenote_section_id(self, onenoteSection_id):
        return DummyDelete(self._response, self._raise_exc)


class DummySectionGroups:
    def __init__(self, response=None, raise_exc: Exception = None):
        self._response = response
        self._raise_exc = raise_exc

    def by_section_group_id(self, sectionGroup_id):
        return DummySections(self._response, self._raise_exc)


class DummyMe:
    def __init__(self, response=None, raise_exc: Exception = None):
        self.onenote = DummyOneNote(response, raise_exc)


class DummyOneNote:
    def __init__(self, response=None, raise_exc: Exception = None):
        self.section_groups = DummySectionGroups(response, raise_exc)


class DummyMSGraphServiceClient:
    def __init__(self, response=None, raise_exc: Exception = None):
        self.me = DummyMe(response, raise_exc)


class DummyMSGraphClient:
    def __init__(self, response=None, raise_exc: Exception = None):
        self._response = response
        self._raise_exc = raise_exc

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return DummyMSGraphServiceClient(self._response, self._raise_exc)


# --- Unit Tests ---

# 1. Basic Test Cases


@pytest.mark.asyncio
async def test_delete_sections_success_basic():
    """Test successful deletion returns success=True and correct data."""
    # Simulate a successful delete operation
    dummy_response = {"deleted": True}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_delete_sections("sgid", "sid")


@pytest.mark.asyncio
async def test_delete_sections_returns_error_dict():
    """Test error response in dict triggers error handling."""
    dummy_response = {"error": {"code": "404", "message": "Not found"}}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_delete_sections("sgid", "sid")


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

    class Response:
        error = "Something went wrong"

    client = DummyMSGraphClient(response=Response())
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_delete_sections("sgid", "sid")


@pytest.mark.asyncio
async def test_delete_sections_returns_none():
    """Test None response triggers error handling."""
    client = DummyMSGraphClient(response=None)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_delete_sections("sgid", "sid")


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

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

    client = DummyMSGraphClient(response=Response())
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_delete_sections("sgid", "sid")


# 2. Edge Test Cases


@pytest.mark.asyncio
async def test_delete_sections_with_all_parameters():
    """Test with all optional parameters set."""
    dummy_response = {"deleted": True}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_delete_sections(
        sectionGroup_id="sgid",
        onenoteSection_id="sid",
        If_Match="etag1",
        select=["id", "name"],
        expand=["parentNotebook"],
        filter="name eq 'Section1'",
        orderby="name desc",
        search="math",
        top=10,
        skip=2,
        headers={"Custom-Header": "Value"},
        custom_param="custom",
    )


@pytest.mark.asyncio
async def test_delete_sections_concurrent_execution():
    """Test concurrent execution of the async function."""
    dummy_response = {"deleted": True}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    # Launch 5 concurrent delete operations
    coros = [
        ds.me_onenote_section_groups_delete_sections(f"sgid{i}", f"sid{i}")
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_delete_sections_exception_handling():
    """Test that an exception in the delete call is handled gracefully."""

    class CustomExc(Exception):
        pass

    client = DummyMSGraphClient(raise_exc=CustomExc("fail!"))
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_delete_sections("sgid", "sid")


@pytest.mark.asyncio
async def test_delete_sections_headers_consistency_level_added():
    """Test that ConsistencyLevel header is added when search is used."""
    # We'll check that no exception is raised and the call completes
    dummy_response = {"deleted": True}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_delete_sections(
        sectionGroup_id="sgid", onenoteSection_id="sid", search="test"
    )


@pytest.mark.asyncio
async def test_delete_sections_with_non_list_select_expand():
    """Test select/expand as single string instead of list."""
    dummy_response = {"deleted": True}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    result = await ds.me_onenote_section_groups_delete_sections(
        sectionGroup_id="sgid",
        onenoteSection_id="sid",
        select="id",
        expand="parentNotebook",
    )


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_delete_sections_many_concurrent():
    """Test with a large number of concurrent executions (50)."""
    dummy_response = {"deleted": True}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    coros = [
        ds.me_onenote_section_groups_delete_sections(f"sgid{i}", f"sid{i}")
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


@pytest.mark.asyncio
async def test_delete_sections_varied_inputs():
    """Test with varied combinations of parameters."""
    dummy_response = {"deleted": True}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    coros = [
        ds.me_onenote_section_groups_delete_sections("sgid", "sid", select=["id"]),
        ds.me_onenote_section_groups_delete_sections(
            "sgid", "sid", expand=["parentNotebook"]
        ),
        ds.me_onenote_section_groups_delete_sections(
            "sgid", "sid", filter="id eq 'sid'"
        ),
        ds.me_onenote_section_groups_delete_sections("sgid", "sid", top=1),
        ds.me_onenote_section_groups_delete_sections("sgid", "sid", skip=0),
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_delete_sections_throughput_small_load():
    """Throughput test: small load (10 concurrent)."""
    dummy_response = {"deleted": True}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    tasks = [
        ds.me_onenote_section_groups_delete_sections(f"sgid{i}", f"sid{i}")
        for i in range(10)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_delete_sections_throughput_medium_load():
    """Throughput test: medium load (100 concurrent)."""
    dummy_response = {"deleted": True}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    tasks = [
        ds.me_onenote_section_groups_delete_sections(f"sgid{i}", f"sid{i}")
        for i in range(100)
    ]
    results = await asyncio.gather(*tasks)


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_delete_sections_throughput_with_errors():
    """Throughput test: mix of success and error responses."""

    class ErrorResponse:
        error = "Simulated error"

    responses = [
        {"deleted": True} if i % 2 == 0 else ErrorResponse() for i in range(20)
    ]
    # Each client gets a different response
    clients = [DummyMSGraphClient(response=resp) for resp in responses]
    datasources = [OneNoteDataSource(client) for client in clients]
    tasks = [
        ds.me_onenote_section_groups_delete_sections(f"sgid{i}", f"sid{i}")
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*tasks)
    for i, result in enumerate(results):
        if i % 2 == 0:
            pass
        else:
            pass


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_delete_sections_throughput_parameter_variation():
    """Throughput test: concurrent calls with varied parameters."""
    dummy_response = {"deleted": True}
    client = DummyMSGraphClient(response=dummy_response)
    ds = OneNoteDataSource(client)
    tasks = [
        ds.me_onenote_section_groups_delete_sections(
            f"sgid{i}",
            f"sid{i}",
            select=["id"] if i % 2 == 0 else None,
            expand=["parentNotebook"] if i % 3 == 0 else None,
            top=i if i % 5 == 0 else None,
        )
        for i in range(30)
    ]
    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  # 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

# -- Dummy Classes to support the function under test --


class OneNoteResponse:
    """Minimal stub for OneNoteResponse to match expected usage."""

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


class DummyDeleteEndpoint:
    """Simulates the delete endpoint for OneNote section deletion."""

    def __init__(self, should_raise=False, return_error=False, response_data=None):
        self.should_raise = should_raise
        self.return_error = return_error
        self.response_data = response_data

    async def delete(self, request_configuration=None):
        # Simulate raising an exception if requested
        if self.should_raise:
            raise RuntimeError("Simulated API failure")
        # Simulate returning an error response if requested
        if self.return_error:
            return {"error": {"code": "NotFound", "message": "Section not found"}}
        # Return custom response data if provided
        if self.response_data is not None:
            return self.response_data
        # Otherwise, simulate successful deletion
        return {"deleted": True}


class DummySections:
    """Simulates the sections property."""

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

    def by_onenote_section_id(self, onenoteSection_id):
        return self.endpoint


class DummySectionGroups:
    """Simulates the section_groups property."""

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

    def by_section_group_id(self, sectionGroup_id):
        return DummySections(self.endpoint)


class DummyMe:
    """Simulates the me property."""

    def __init__(self, endpoint):
        self.endpoint = endpoint
        self.onenote = self

    @property
    def section_groups(self):
        return DummySectionGroups(self.endpoint)


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

    def __init__(self, endpoint):
        self.me = DummyMe(endpoint)

    def get_ms_graph_service_client(self):
        return self


# -- Dummy MSGraphClient for __init__ --
class MSGraphClient:
    def __init__(self, client):
        self.client = client

    def get_client(self):
        return self.client


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

# Basic Test Cases


@pytest.mark.asyncio
async def test_delete_sections_basic_success():
    """Test basic successful deletion."""
    endpoint = DummyDeleteEndpoint()
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    resp = await ds.me_onenote_section_groups_delete_sections("sg1", "sec1")


@pytest.mark.asyncio
async def test_delete_sections_basic_with_select_expand():
    """Test with select and expand options."""
    endpoint = DummyDeleteEndpoint()
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    resp = await ds.me_onenote_section_groups_delete_sections(
        "sg2", "sec2", select=["id", "name"], expand=["pages"]
    )


@pytest.mark.asyncio
async def test_delete_sections_basic_with_headers():
    """Test with custom headers."""
    endpoint = DummyDeleteEndpoint()
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    resp = await ds.me_onenote_section_groups_delete_sections(
        "sg3", "sec3", headers={"Authorization": "Bearer xyz"}
    )


# Edge Test Cases


@pytest.mark.asyncio
async def test_delete_sections_error_response():
    """Test when API returns an error response dict."""
    endpoint = DummyDeleteEndpoint(return_error=True)
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    resp = await ds.me_onenote_section_groups_delete_sections("sg4", "sec4")


@pytest.mark.asyncio
async def test_delete_sections_exception_handling():
    """Test when API raises an exception."""
    endpoint = DummyDeleteEndpoint(should_raise=True)
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    resp = await ds.me_onenote_section_groups_delete_sections("sg5", "sec5")


@pytest.mark.asyncio
async def test_delete_sections_none_response_handling():
    """Test when API returns None (empty response)."""
    endpoint = DummyDeleteEndpoint(response_data=None)
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    # Patch endpoint to return None explicitly
    endpoint.delete = lambda request_configuration=None: asyncio.sleep(0, result=None)
    resp = await ds.me_onenote_section_groups_delete_sections("sg6", "sec6")


@pytest.mark.asyncio
async def test_delete_sections_with_search_and_consistency_header():
    """Test with search parameter, which should add ConsistencyLevel header."""
    endpoint = DummyDeleteEndpoint()
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    resp = await ds.me_onenote_section_groups_delete_sections(
        "sg7", "sec7", search="test search"
    )


@pytest.mark.asyncio
async def test_delete_sections_concurrent_calls():
    """Test concurrent deletion calls for async correctness."""
    endpoint = DummyDeleteEndpoint()
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    # Run 5 concurrent deletions
    results = await asyncio.gather(
        *[
            ds.me_onenote_section_groups_delete_sections(f"sg{i}", f"sec{i}")
            for i in range(5)
        ]
    )
    for resp in results:
        pass


# Large Scale Test Cases


@pytest.mark.asyncio
async def test_delete_sections_large_scale_concurrent():
    """Test large scale concurrent deletions (up to 50)."""
    endpoint = DummyDeleteEndpoint()
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    # 50 concurrent deletions
    results = await asyncio.gather(
        *[
            ds.me_onenote_section_groups_delete_sections(f"sg{i}", f"sec{i}")
            for i in range(50)
        ]
    )
    for resp in results:
        pass


@pytest.mark.asyncio
async def test_delete_sections_large_scale_with_errors():
    """Test large scale concurrent deletions with some errors."""
    endpoints = [DummyDeleteEndpoint(return_error=(i % 10 == 0)) for i in range(30)]
    clients = [MSGraphClient(DummyClient(e)) for e in endpoints]
    datasources = [OneNoteDataSource(c) for c in clients]
    coros = [
        ds.me_onenote_section_groups_delete_sections(f"sg{i}", f"sec{i}")
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*coros)
    for i, resp in enumerate(results):
        if i % 10 == 0:
            pass
        else:
            pass


# Throughput Test Cases


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_delete_sections_throughput_small_load():
    """Throughput test: small load (10 deletions)."""
    endpoint = DummyDeleteEndpoint()
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    coros = [
        ds.me_onenote_section_groups_delete_sections(f"sg{i}", f"sec{i}")
        for i in range(10)
    ]
    results = await asyncio.gather(*coros)
    for resp in results:
        pass


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_delete_sections_throughput_medium_load():
    """Throughput test: medium load (100 deletions)."""
    endpoint = DummyDeleteEndpoint()
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    coros = [
        ds.me_onenote_section_groups_delete_sections(f"sg{i}", f"sec{i}")
        for i in range(100)
    ]
    results = await asyncio.gather(*coros)
    for resp in results:
        pass


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_delete_sections_throughput_high_volume():
    """Throughput test: high volume (500 deletions)."""
    endpoint = DummyDeleteEndpoint()
    client = MSGraphClient(DummyClient(endpoint))
    ds = OneNoteDataSource(client)
    coros = [
        ds.me_onenote_section_groups_delete_sections(f"sg{i}", f"sec{i}")
        for i in range(500)
    ]
    results = await asyncio.gather(*coros)
    for resp in results:
        pass


@pytest.mark.asyncio
async def test_OneNoteDataSource_me_onenote_section_groups_delete_sections_throughput_mixed_load():
    """Throughput test: mixed load with errors (100 deletions, 20 errors)."""
    endpoints = [DummyDeleteEndpoint(return_error=(i % 5 == 0)) for i in range(100)]
    clients = [MSGraphClient(DummyClient(e)) for e in endpoints]
    datasources = [OneNoteDataSource(c) for c in clients]
    coros = [
        ds.me_onenote_section_groups_delete_sections(f"sg{i}", f"sec{i}")
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*coros)
    error_count = sum(1 for resp in results if not resp.success)
    for i, resp 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.

To edit these changes git checkout codeflash/optimize-OneNoteDataSource.me_onenote_section_groups_delete_sections-mjfykts4 and push.

Codeflash Static Badge

The optimized code achieves an **11% runtime improvement** and **1.3% throughput increase** by reducing object instantiation overhead in the query parameter handling logic.

**Key optimization applied:**
- **Replaced double `RequestConfiguration()` instantiation with a single instance** - The original code created one `RequestConfiguration` for `query_params` and another for `config`, then always assigned the first to the second regardless of whether any parameters were set.
- **Used a lightweight dictionary for parameter collection** - Instead of immediately creating a `RequestConfiguration` object and setting its attributes, parameters are first collected in a plain dictionary (`query_params = {}`).
- **Conditional assignment of query parameters** - The `config.query_parameters` assignment only happens when there are actual parameters to set (`if query_params:`), avoiding unnecessary object attribute assignments.

**Why this leads to speedup:**
- **Reduced object creation overhead** - `RequestConfiguration()` instantiation involves class initialization overhead that's eliminated for the first object
- **Fewer attribute assignments** - Dictionary operations (`query_params["key"] = value`) are faster than object attribute assignments (`query_params.key = value`)
- **Conditional processing** - The optimization only assigns `query_parameters` when needed, avoiding unnecessary work when no query parameters are present

**Performance characteristics:**
- The line profiler shows the original `RequestConfiguration()` call took 1.11ms vs 0.26ms for the dictionary initialization
- Most effective for calls with few or no query parameters (which appears to be the majority based on test patterns)
- The async nature means this optimization compounds across concurrent operations, explaining the throughput improvement

This optimization is particularly valuable for high-frequency OneNote API operations where query parameters are often sparse or absent.
@codeflash-ai codeflash-ai bot requested a review from mashraf-222 December 21, 2025 16:45
@codeflash-ai codeflash-ai bot added ⚡️ codeflash Optimization PR opened by Codeflash AI 🎯 Quality: Medium Optimization Quality according to Codeflash labels Dec 21, 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: Medium Optimization Quality according to Codeflash

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant