⚡️ Speed up method OneNoteDataSource.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook by 7%

[codeflash-ai]

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

⏱️ Runtime : 800 microseconds → 746 microseconds (best of 58 runs)

📝 Explanation and details

The optimized code achieves a 7% runtime improvement through two key optimizations that reduce unnecessary object creation and attribute assignments:

1. Response Handling Optimization (_handle_onenote_response)

• Early returns for error cases: Instead of setting success=False and error_msg variables then constructing the response at the end, the code immediately returns OneNoteResponse objects when errors are detected
• Eliminates redundant variable assignments: Removes the success=True and error_msg=None initialization since the success path now directly returns without intermediate variables
• Reduces branching overhead: The optimized flow has fewer conditional branches and variable assignments in the common execution paths

2. Query Parameters Optimization

• Conditional object creation: Only creates RequestConfiguration() for query parameters when at least one parameter is actually provided, avoiding object instantiation in the common case where no query parameters are needed
• Lazy assignment: Uses if query_params: check before assigning config.query_parameters, eliminating unnecessary attribute assignments
• Safer attribute access: Replaces if not config.headers: with if not getattr(config, "headers", None): to avoid potential attribute errors

Performance Impact:
The line profiler shows the most significant gains in the response handling method, where the total time decreased from 736μs to 677μs. The main async function shows reduced object creation overhead, particularly beneficial when no query parameters are provided (the common case in most OneNote API calls).

Test Case Benefits:
These optimizations are most effective for:

• Basic operations with minimal parameters (tests show consistent improvements)
• High-throughput concurrent operations where object creation overhead compounds
• Error handling scenarios where early returns avoid unnecessary processing

The optimizations maintain identical functionality while reducing CPU cycles through smarter object lifecycle management.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio

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

# --- Minimal stubs for dependencies ---


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


# --- Minimal MSGraphClient and chainable client stubs for testing ---


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

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


class SectionsStub:
    def __init__(self, copy_to_notebook):
        self.copy_to_notebook = copy_to_notebook

    def by_onenote_section_id(self, onenote_section_id):
        return self


class NotebooksStub:
    def __init__(self, sections):
        self.sections = sections

    def by_notebook_id(self, notebook_id):
        return self


class OneNoteStub:
    def __init__(self, notebooks):
        self.notebooks = notebooks


class GroupsStub:
    def __init__(self, onenote):
        self.onenote = onenote

    def by_group_id(self, group_id):
        return self


class ClientStub:
    def __init__(self, response=None, raise_exc=None):
        # Build the chain
        self.groups = GroupsStub(
            OneNoteStub(
                NotebooksStub(
                    SectionsStub(
                        CopyToNotebookStub(response=response, raise_exc=raise_exc)
                    )
                )
            )
        )
        self.me = True  # For __init__ check

    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_copy_to_notebook_basic_success():
    """Test basic successful copy operation with minimal required arguments."""

    # Simulate a successful response object
    class Resp:
        pass

    resp = Resp()
    ds = OneNoteDataSource(ClientStub(response=resp))
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g1", notebook_id="n1", onenoteSection_id="s1"
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_basic_with_options():
    """Test with select, expand, filter, orderby, search, top, skip, headers, and request_body."""
    resp = {"id": "section123"}
    ds = OneNoteDataSource(ClientStub(response=resp))
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g2",
        notebook_id="n2",
        onenoteSection_id="s2",
        select=["id", "name"],
        expand=["pages"],
        filter="name eq 'Test'",
        orderby="name",
        search="test",
        top=10,
        skip=5,
        request_body={"targetNotebookId": "n3"},
        headers={"Authorization": "Bearer token"},
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_basic_none_response():
    """Test when the response is None (should return OneNoteResponse with success=False)."""
    ds = OneNoteDataSource(ClientStub(response=None))
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g3", notebook_id="n3", onenoteSection_id="s3"
    )


# 2. Edge Test Cases


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

    class Resp:
        error = "Something went wrong"

    ds = OneNoteDataSource(ClientStub(response=Resp()))
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g4", notebook_id="n4", onenoteSection_id="s4"
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_response_with_error_dict():
    """Test when the response is a dict with an 'error' key (dict value)."""
    error_dict = {"error": {"code": "BadRequest", "message": "Bad request"}}
    ds = OneNoteDataSource(ClientStub(response=error_dict))
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g5", notebook_id="n5", onenoteSection_id="s5"
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_response_with_error_dict_str():
    """Test when the response is a dict with an 'error' key (str value)."""
    error_dict = {"error": "API error"}
    ds = OneNoteDataSource(ClientStub(response=error_dict))
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g6", notebook_id="n6", onenoteSection_id="s6"
    )


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

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

    ds = OneNoteDataSource(ClientStub(response=Resp()))
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g7", notebook_id="n7", onenoteSection_id="s7"
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_exception_handling():
    """Test that an exception in the post method is caught and returned as error."""

    class CustomExc(Exception):
        pass

    ds = OneNoteDataSource(ClientStub(raise_exc=CustomExc("fail!")))
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g8", notebook_id="n8", onenoteSection_id="s8"
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_invalid_client():
    """Test that __init__ raises if client does not have 'me' attribute."""

    class BadClient:
        def get_client(self):
            return self

        def get_ms_graph_service_client(self):
            return self

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


@pytest.mark.asyncio
async def test_copy_to_notebook_concurrent_execution():
    """Test concurrent execution of multiple copy operations."""
    resp1, resp2 = {"id": "a"}, {"id": "b"}
    ds1 = OneNoteDataSource(ClientStub(response=resp1))
    ds2 = OneNoteDataSource(ClientStub(response=resp2))
    # Run both concurrently
    results = await asyncio.gather(
        ds1.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id="g9", notebook_id="n9", onenoteSection_id="s9"
        ),
        ds2.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id="g10", notebook_id="n10", onenoteSection_id="s10"
        ),
    )


# 3. Large Scale Test Cases


@pytest.mark.asyncio
async def test_copy_to_notebook_large_scale_concurrent():
    """Test large number of concurrent copy operations (up to 50)."""
    responses = [{"id": f"section_{i}"} for i in range(50)]
    data_sources = [OneNoteDataSource(ClientStub(response=resp)) for resp in responses]
    coros = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id=f"g{i}", notebook_id=f"n{i}", onenoteSection_id=f"s{i}"
        )
        for i, ds in enumerate(data_sources)
    ]
    results = await asyncio.gather(*coros)
    for i, r in enumerate(results):
        pass


# 4. Throughput Test Cases


@pytest.mark.asyncio
async def test_copy_to_notebook_throughput_small_load():
    """Throughput test: small batch of 5 concurrent requests."""
    responses = [{"id": f"small_{i}"} for i in range(5)]
    data_sources = [OneNoteDataSource(ClientStub(response=resp)) for resp in responses]
    coros = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id=f"gs{i}", notebook_id=f"ns{i}", onenoteSection_id=f"ss{i}"
        )
        for i, ds in enumerate(data_sources)
    ]
    results = await asyncio.gather(*coros)
    for i, r in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_copy_to_notebook_throughput_medium_load():
    """Throughput test: medium batch of 20 concurrent requests."""
    responses = [{"id": f"med_{i}"} for i in range(20)]
    data_sources = [OneNoteDataSource(ClientStub(response=resp)) for resp in responses]
    coros = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id=f"gm{i}", notebook_id=f"nm{i}", onenoteSection_id=f"sm{i}"
        )
        for i, ds in enumerate(data_sources)
    ]
    results = await asyncio.gather(*coros)
    for i, r in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_copy_to_notebook_throughput_large_load():
    """Throughput test: large batch of 100 concurrent requests."""
    responses = [{"id": f"large_{i}"} for i in range(100)]
    data_sources = [OneNoteDataSource(ClientStub(response=resp)) for resp in responses]
    coros = [
        ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id=f"gl{i}", notebook_id=f"nl{i}", onenoteSection_id=f"sl{i}"
        )
        for i, ds in enumerate(data_sources)
    ]
    results = await asyncio.gather(*coros)
    for i, r in enumerate(results):
        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  # used to run async functions

# ---- Logger stub ----
from typing import Any, Optional

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

# ---- Minimal stubs and helpers for the test environment ----


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


class DummyCopyToNotebook:
    """Simulates the .post() async method for OneNote section copy."""

    def __init__(self, response):
        self._response = response

    async def post(self, body=None, request_configuration=None):
        # Simulate async call returning a response object/dict
        if isinstance(self._response, Exception):
            raise self._response
        return self._response


class DummySections:
    def __init__(self, response):
        self._response = response

    def by_onenote_section_id(self, onenote_section_id):
        return self

    @property
    def copy_to_notebook(self):
        return DummyCopyToNotebook(self._response)


class DummyNotebooks:
    def __init__(self, response):
        self._response = response

    def by_notebook_id(self, notebook_id):
        return DummySections(self._response)


class DummyOnenote:
    def __init__(self, response):
        self._response = response

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


class DummyGroups:
    def __init__(self, response):
        self._response = response

    def by_group_id(self, group_id):
        return DummyOnenote(self._response)


class DummyClient:
    """Simulates the .groups property chain for the OneNoteDataSource."""

    def __init__(self, response):
        self._response = response

    @property
    def groups(self):
        return DummyGroups(self._response)

    @property
    def me(self):
        return True  # Used for OneNoteDataSource __init__ check


class DummyMSGraphClient:
    """Simulates the MSGraphClient.get_client().get_ms_graph_service_client()."""

    def __init__(self, response):
        self._response = response

    def get_client(self):
        return self

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


# ---- UNIT TESTS ----

# Basic Test Cases


@pytest.mark.asyncio
async def test_copy_to_notebook_returns_successful_response():
    """Test that the function returns a successful OneNoteResponse with expected data."""
    # Simulate a successful response object
    response_data = {"id": "section123", "status": "copied"}
    msgraph_client = DummyMSGraphClient(response_data)
    ds = OneNoteDataSource(msgraph_client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g1", notebook_id="n1", onenoteSection_id="s1"
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_returns_error_response_dict():
    """Test that a dict with 'error' key is handled as error."""
    response_data = {"error": {"code": "BadRequest", "message": "Invalid section id"}}
    msgraph_client = DummyMSGraphClient(response_data)
    ds = OneNoteDataSource(msgraph_client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g2", notebook_id="n2", onenoteSection_id="bad"
    )


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

    class ErrorObj:
        error = "Some error occurred"

    msgraph_client = DummyMSGraphClient(ErrorObj())
    ds = OneNoteDataSource(msgraph_client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g3", notebook_id="n3", onenoteSection_id="s3"
    )


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

    class ErrObj:
        code = "404"
        message = "Section not found"

    msgraph_client = DummyMSGraphClient(ErrObj())
    ds = OneNoteDataSource(msgraph_client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g4", notebook_id="n4", onenoteSection_id="s4"
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_handles_none_response():
    """Test that a None response is handled with correct error."""
    msgraph_client = DummyMSGraphClient(None)
    ds = OneNoteDataSource(msgraph_client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g5", notebook_id="n5", onenoteSection_id="s5"
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_with_all_parameters():
    """Test that all parameters are accepted and processed."""
    response_data = {"id": "section456", "status": "copied"}
    msgraph_client = DummyMSGraphClient(response_data)
    ds = OneNoteDataSource(msgraph_client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g6",
        notebook_id="n6",
        onenoteSection_id="s6",
        select=["id", "status"],
        expand=["pages"],
        filter="status eq 'active'",
        orderby="createdDateTime desc",
        search="biology",
        top=5,
        skip=1,
        request_body={"name": "Copy"},
        headers={"Authorization": "Bearer token"},
        custom_param="custom",
    )


# Edge Test Cases


@pytest.mark.asyncio
async def test_copy_to_notebook_concurrent_execution():
    """Test concurrent execution of multiple copy operations with different results."""
    responses = [
        {"id": "s1", "status": "copied"},
        {"error": {"code": "Conflict", "message": "Already exists"}},
        {"id": "s3", "status": "copied"},
    ]
    msgraph_clients = [DummyMSGraphClient(r) for r in responses]
    ds_list = [OneNoteDataSource(c) for c in msgraph_clients]

    async def run(ds, gid, nid, sid):
        return await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id=gid, notebook_id=nid, onenoteSection_id=sid
        )

    # Run all concurrently
    results = await asyncio.gather(
        run(ds_list[0], "g1", "n1", "s1"),
        run(ds_list[1], "g2", "n2", "s2"),
        run(ds_list[2], "g3", "n3", "s3"),
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_handles_exception_in_post():
    """Test that an exception in the post() call results in a failed OneNoteResponse."""
    msgraph_client = DummyMSGraphClient(RuntimeError("Network error"))
    ds = OneNoteDataSource(msgraph_client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g7", notebook_id="n7", onenoteSection_id="s7"
    )


@pytest.mark.asyncio
async def test_copy_to_notebook_handles_invalid_client():
    """Test that __init__ raises if client does not have 'me' attribute."""

    class BadClient:
        def get_client(self):
            return self

        def get_ms_graph_service_client(self):
            return object()

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


@pytest.mark.asyncio
async def test_copy_to_notebook_with_search_adds_consistency_header():
    """Test that ConsistencyLevel header is set when search is used."""

    class HeaderCheckCopyToNotebook(DummyCopyToNotebook):
        async def post(self, body=None, request_configuration=None):
            # Check if ConsistencyLevel header is set
            headers = getattr(request_configuration, "headers", None)
            return {"id": "section999", "status": "copied"}

    class HeaderCheckSections(DummySections):
        @property
        def copy_to_notebook(self):
            return HeaderCheckCopyToNotebook(self._response)

    class HeaderCheckNotebooks(DummyNotebooks):
        def by_notebook_id(self, notebook_id):
            return HeaderCheckSections(self._response)

    class HeaderCheckOnenote(DummyOnenote):
        @property
        def notebooks(self):
            return HeaderCheckNotebooks(self._response)

    class HeaderCheckGroups(DummyGroups):
        def by_group_id(self, group_id):
            return HeaderCheckOnenote(self._response)

    class HeaderCheckClient(DummyClient):
        @property
        def groups(self):
            return HeaderCheckGroups(self._response)

    class HeaderCheckMSGraphClient(DummyMSGraphClient):
        def get_ms_graph_service_client(self):
            return HeaderCheckClient(self._response)

    msgraph_client = HeaderCheckMSGraphClient({"id": "section999", "status": "copied"})
    ds = OneNoteDataSource(msgraph_client)
    result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
        group_id="g8", notebook_id="n8", onenoteSection_id="s8", search="test"
    )


# Large Scale Test Cases


@pytest.mark.asyncio
async def test_copy_to_notebook_large_scale_concurrent():
    """Test 20 concurrent copy operations for scalability and async correctness."""
    N = 20
    responses = [{"id": f"s{i}", "status": "copied"} for i in range(N)]
    msgraph_clients = [DummyMSGraphClient(r) for r in responses]
    ds_list = [OneNoteDataSource(c) for c in msgraph_clients]

    async def run(ds, i):
        return await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id=f"g{i}", notebook_id=f"n{i}", onenoteSection_id=f"s{i}"
        )

    results = await asyncio.gather(*[run(ds, i) for i, ds in enumerate(ds_list)])
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_copy_to_notebook_large_scale_with_errors():
    """Test 10 concurrent operations with mixed success and error responses."""
    N = 10
    responses = []
    for i in range(N):
        if i % 2 == 0:
            responses.append({"id": f"s{i}", "status": "copied"})
        else:
            responses.append({"error": {"code": "409", "message": "Duplicate"}})
    msgraph_clients = [DummyMSGraphClient(r) for r in responses]
    ds_list = [OneNoteDataSource(c) for c in msgraph_clients]

    async def run(ds, i):
        return await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id=f"g{i}", notebook_id=f"n{i}", onenoteSection_id=f"s{i}"
        )

    results = await asyncio.gather(*[run(ds, i) for i, ds in enumerate(ds_list)])
    for i, result in enumerate(results):
        if i % 2 == 0:
            pass
        else:
            pass


# Throughput Test Cases


@pytest.mark.asyncio
async def test_copy_to_notebook_throughput_small_load():
    """Throughput: Test 5 sequential copy operations complete quickly and correctly."""
    responses = [{"id": f"s{i}", "status": "copied"} for i in range(5)]
    msgraph_clients = [DummyMSGraphClient(r) for r in responses]
    ds_list = [OneNoteDataSource(c) for c in msgraph_clients]
    for i, ds in enumerate(ds_list):
        result = await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id=f"g{i}", notebook_id=f"n{i}", onenoteSection_id=f"s{i}"
        )


@pytest.mark.asyncio
async def test_copy_to_notebook_throughput_medium_concurrent():
    """Throughput: Test 50 concurrent copy operations for async throughput."""
    N = 50
    responses = [{"id": f"s{i}", "status": "copied"} for i in range(N)]
    msgraph_clients = [DummyMSGraphClient(r) for r in responses]
    ds_list = [OneNoteDataSource(c) for c in msgraph_clients]

    async def run(ds, i):
        return await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id=f"g{i}", notebook_id=f"n{i}", onenoteSection_id=f"s{i}"
        )

    results = await asyncio.gather(*[run(ds, i) for i, ds in enumerate(ds_list)])
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_copy_to_notebook_throughput_mixed_load():
    """Throughput: Test 30 concurrent operations with 10 errors and 20 successes."""
    N = 30
    responses = []
    for i in range(N):
        if i % 3 == 0:
            responses.append({"error": {"code": "500", "message": "Internal Error"}})
        else:
            responses.append({"id": f"s{i}", "status": "copied"})
    msgraph_clients = [DummyMSGraphClient(r) for r in responses]
    ds_list = [OneNoteDataSource(c) for c in msgraph_clients]

    async def run(ds, i):
        return await ds.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook(
            group_id=f"g{i}", notebook_id=f"n{i}", onenoteSection_id=f"s{i}"
        )

    results = await asyncio.gather(*[run(ds, i) for i, ds in enumerate(ds_list)])
    for i, result in enumerate(results):
        if i % 3 == 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.groups_group_onenote_notebooks_notebook_sections_onenote_section_copy_to_notebook-mjb1jvuh and push.