⚡️ Speed up method OneNoteDataSource.employee_experience_learning_providers_learning_contents_get_by_external_id by 51%

[codeflash-ai]

📄 51% (0.51x) speedup for OneNoteDataSource.employee_experience_learning_providers_learning_contents_get_by_external_id in backend/python/app/sources/external/microsoft/one_note/one_note.py

⏱️ Runtime : 807 microseconds → 534 microseconds (best of 5 runs)

📝 Explanation and details

The optimized code achieves a 51% runtime improvement through several key optimizations that reduce object creation overhead and eliminate unnecessary operations:

Primary Optimizations:

1. Conditional RequestConfiguration Creation: The original code always creates a RequestConfiguration() object even when no query parameters are provided. The optimized version uses any([...]) to check if query parameters are needed first, only creating the object when necessary. This eliminates ~196 unnecessary object creations in typical usage patterns.

2. Early Return Pattern in Error Handling: The _handle_onenote_response method now uses immediate returns for error conditions instead of setting variables and falling through to a single return statement. This reduces attribute assignments and eliminates the need to track success and error_msg variables through the entire method.

3. Defensive Header Copying: Headers are only copied when provided (headers.copy()), preventing unnecessary dictionary operations when headers are None.

4. Query Builder Pattern: The API call is restructured to build the query chain step-by-step and store it in a variable before making the actual call, potentially improving readability and reducing repeated attribute access.

Performance Impact:
The line profiler shows the most significant improvements in the main async method, where total execution time dropped from 3.79ms to 1.91ms. The optimization is particularly effective for the common case where most query parameters are None, which appears to represent ~98.5% of test calls based on the profiler data.

Throughput Analysis:
While throughput remains constant at 995 operations/second, the 51% runtime reduction means each individual operation completes faster, which is valuable for latency-sensitive scenarios and reduces resource consumption per operation.

Test Case Benefits:
The optimization is most effective for basic API calls without complex query parameters (like test_basic_success), which represent the majority of real-world usage patterns. More complex calls with many parameters still benefit but to a lesser degree.

✅ Correctness verification report:

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


class OneNoteResponse:
    """A minimal OneNoteResponse class for testing."""

    def __init__(self, success: bool, data=None, error=None):
        self.success = success
        self.data = data
        self.error = error


class DummyLearningContents:
    """Simulates the .learning_contents(external_id=...) method chain."""

    def __init__(self, learningProvider_id, parent):
        self.learningProvider_id = learningProvider_id
        self.parent = parent

    def learning_contents(self, external_id):
        self.external_id = external_id
        return self

    async def get(self, request_configuration=None):
        # Simulate different responses based on input
        # The external_id is passed as "{external_id}" so we extract the value
        ext_id = self.external_id
        # Remove curly braces if present
        if ext_id.startswith("{") and ext_id.endswith("}"):
            ext_id = ext_id[1:-1]
        # Simulate different behaviors for test cases
        if self.learningProvider_id == "raise" or ext_id == "raise":
            raise Exception("Simulated API error")
        if self.learningProvider_id == "none" or ext_id == "none":
            return None
        # Simulate a normal response
        data = {
            "learningProvider_id": self.learningProvider_id,
            "external_id": ext_id,
            "select": getattr(request_configuration.query_parameters, "select", None)
            if request_configuration and request_configuration.query_parameters
            else None,
            "expand": getattr(request_configuration.query_parameters, "expand", None)
            if request_configuration and request_configuration.query_parameters
            else None,
            "headers": getattr(request_configuration, "headers", None)
            if request_configuration
            else None,
            "filter": getattr(request_configuration.query_parameters, "filter", None)
            if request_configuration and request_configuration.query_parameters
            else None,
            "orderby": getattr(request_configuration.query_parameters, "orderby", None)
            if request_configuration and request_configuration.query_parameters
            else None,
            "search": getattr(request_configuration.query_parameters, "search", None)
            if request_configuration and request_configuration.query_parameters
            else None,
            "top": getattr(request_configuration.query_parameters, "top", None)
            if request_configuration and request_configuration.query_parameters
            else None,
            "skip": getattr(request_configuration.query_parameters, "skip", None)
            if request_configuration and request_configuration.query_parameters
            else None,
        }
        # Simulate error field for edge case
        if ext_id == "error":
            return {"error": {"code": "404", "message": "Not Found"}}
        return data


class DummyLearningProviders:
    """Simulates the .by_learningProvider_id(...) method chain."""

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

    def by_learningProvider_id(self, learningProvider_id):
        return DummyLearningContents(learningProvider_id, self.parent)


class DummyEmployeeExperience:
    """Simulates the .employee_experience property."""

    def __init__(self, parent):
        self.parent = parent
        self.learning_providers = DummyLearningProviders(self.parent)


class DummyClient:
    """Simulates the Microsoft Graph SDK client for OneNote."""

    def __init__(self):
        self.me = True  # To pass the hasattr check
        self.employee_experience = DummyEmployeeExperience(self)


class DummyMSGraphClient:
    """Simulates the MSGraphClient wrapper."""

    def __init__(self):
        self._client = DummyClient()

    def get_client(self):
        return self._client


# --- Patch DummyClient to provide get_ms_graph_service_client for __init__ compatibility ---
def get_ms_graph_service_client(self):
    return self


DummyClient.get_ms_graph_service_client = get_ms_graph_service_client

# --- TESTS ---


@pytest.mark.asyncio
async def test_basic_success():
    """Basic test: Normal successful call returns expected OneNoteResponse."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    result = await ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
        "provider1", "external1"
    )


@pytest.mark.asyncio
async def test_basic_select_and_expand():
    """Test select and expand parameters are passed and reflected in the response data."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    result = await ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
        "provider2", "external2", select=["id", "name"], expand=["pages"]
    )


@pytest.mark.asyncio
async def test_basic_headers_and_filter():
    """Test headers and filter parameters."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    custom_headers = {"Authorization": "Bearer fake"}
    result = await ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
        "provider3", "external3", headers=custom_headers, filter="name eq 'abc'"
    )


@pytest.mark.asyncio
async def test_edge_case_none_response():
    """Test when API returns None (simulates missing resource)."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    result = await ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
        "none", "external4"
    )


@pytest.mark.asyncio
async def test_edge_case_api_error():
    """Test when API raises an exception (simulates network or server error)."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    result = await ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
        "raise", "external5"
    )


@pytest.mark.asyncio
async def test_edge_case_error_field_in_response():
    """Test when API response contains an error field (simulates API-level error)."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    result = await ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
        "provider6", "error"
    )


@pytest.mark.asyncio
async def test_edge_case_all_parameters():
    """Test all parameters are accepted and handled."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    result = await ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
        "provider7",
        "external7",
        select=["id"],
        expand=["pages"],
        filter="id ne null",
        orderby="name asc",
        search="python",
        top=5,
        skip=2,
        headers={"Custom": "Header"},
    )


@pytest.mark.asyncio
async def test_concurrent_execution():
    """Test concurrent calls with different parameters."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    coros = [
        ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
            f"provider{i}", f"external{i}"
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_concurrent_with_error():
    """Test concurrent calls where one raises an error."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    coros = [
        ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
            "provider8", "external8"
        ),
        ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
            "raise", "external9"
        ),
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_large_scale_concurrent():
    """Test 20 concurrent calls for scalability."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    coros = [
        ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
            f"provider{i}", f"external{i}"
        )
        for i in range(20)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass


@pytest.mark.asyncio
async def test_employee_experience_learning_providers_learning_contents_get_by_external_id_throughput_small_load():
    """Throughput: Test 5 concurrent calls complete quickly."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    coros = [
        ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
            f"p{i}", f"e{i}"
        )
        for i in range(5)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_employee_experience_learning_providers_learning_contents_get_by_external_id_throughput_medium_load():
    """Throughput: Test 50 concurrent calls."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    coros = [
        ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
            f"p{i}", f"e{i}"
        )
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)


@pytest.mark.asyncio
async def test_employee_experience_learning_providers_learning_contents_get_by_external_id_throughput_with_errors():
    """Throughput: Mix of successful and error calls."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    coros = []
    for i in range(10):
        if i % 3 == 0:
            coros.append(
                ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
                    "raise", f"e{i}"
                )
            )
        else:
            coros.append(
                ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
                    f"p{i}", f"e{i}"
                )
            )
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        if i % 3 == 0:
            pass
        else:
            pass


@pytest.mark.asyncio
async def test_employee_experience_learning_providers_learning_contents_get_by_external_id_throughput_high_volume():
    """Throughput: High volume test (100 concurrent calls, all succeed)."""
    client = DummyMSGraphClient()
    ds = OneNoteDataSource(client)
    coros = [
        ds.employee_experience_learning_providers_learning_contents_get_by_external_id(
            f"provider{i}", f"external{i}"
        )
        for i in range(100)
    ]
    results = await asyncio.gather(*coros)


# 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_learning_contents_get_by_external_id-mjgxt64r and push.