⚡️ Speed up method UsersGroupsDataSource.domain_dns_records_domain_dns_record_list_domain_dns_record by 130%

[codeflash-ai]

📄 130% (1.30x) speedup for UsersGroupsDataSource.domain_dns_records_domain_dns_record_list_domain_dns_record in backend/python/app/sources/external/microsoft/users_groups/users_groups.py

⏱️ Runtime : 5.26 milliseconds → 2.29 milliseconds (best of 47 runs)

📝 Explanation and details

The optimized code achieves a 130% speedup by streamlining the query parameter configuration process in the domain_dns_records_domain_dns_record_list_domain_dns_record method.

Key optimizations applied:

1. Batched parameter collection: Instead of creating a RequestConfiguration() object immediately and setting properties one by one, the code now collects all non-None parameters into a dictionary (qp_kwargs) first.

2. Reduced object instantiation: The original code created a RequestConfiguration() object early and then accessed its query_parameters property multiple times. The optimized version defers this creation and uses RequestConfiguration().query_parameters only once.

3. Efficient parameter assignment: Rather than multiple individual property assignments (e.g., query_params.select = select), the optimized code uses a loop with setattr() to batch-assign all collected parameters.

Why this leads to speedup:

• Fewer object instantiations: Reduces memory allocation overhead by minimizing intermediate object creation
• Reduced property access overhead: The original code accessed query_params properties in separate conditional blocks, while the optimized version batches these operations
• More efficient parameter handling: The dictionary-based collection and loop-based assignment pattern is more efficient than individual conditional assignments

Test case performance patterns:
The optimization is particularly effective for test cases with multiple query parameters (select, expand, filter, orderby, top, skip), as seen in the "varied parameters" and "concurrent execution" tests. The throughput improvement of 0.5% demonstrates consistent performance gains across different workload patterns, with the most benefit occurring when multiple query parameters are present.

The import reordering (moving imports to alphabetical order) provides minimal performance benefit but improves code organization without affecting runtime behavior.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
# Function under test (EXACT COPY, DO NOT MODIFY)
import logging
from unittest.mock import AsyncMock, MagicMock

import pytest  # used for our unit tests
from app.sources.external.microsoft.users_groups.users_groups import \
    UsersGroupsDataSource


# Dummy UsersGroupsResponse class for test purposes
class UsersGroupsResponse:
    def __init__(self, success, data=None, error=None):
        self.success = success
        self.data = data
        self.error = error

# -------------------------------
# Fixtures and Mocks for Testing
# -------------------------------

@pytest.fixture
def mock_client():
    """
    Fixture to create a mock client with the required async method.
    """
    # Mock the .domain_dns_records.get() async method
    mock_domain_dns_records = MagicMock()
    mock_domain_dns_records.get = AsyncMock()
    # Mock the ms_graph_service_client to have .domain_dns_records
    mock_ms_graph_service_client = MagicMock()
    mock_ms_graph_service_client.domain_dns_records = mock_domain_dns_records
    # Mock .get_ms_graph_service_client() to return our service client
    mock_client_obj = MagicMock()
    mock_client_obj.get_ms_graph_service_client.return_value = mock_ms_graph_service_client
    # Mock .get_client() to return our client_obj
    mock_client_wrapper = MagicMock()
    mock_client_wrapper.get_client.return_value = mock_client_obj
    return mock_client_wrapper

@pytest.fixture
def datasource(mock_client):
    """
    Fixture to create a UsersGroupsDataSource using the mock client.
    """
    return UsersGroupsDataSource(mock_client)

# -------------------------------
# 1. Basic Test Cases
# -------------------------------

@pytest.mark.asyncio
async def test_basic_success(datasource):
    """
    Test that the function returns a successful UsersGroupsResponse with data when the API call succeeds.
    """
    # Arrange: Set up the mock to return a simple dict
    expected_data = {"records": [{"id": "1"}, {"id": "2"}]}
    datasource.client.domain_dns_records.get.return_value = expected_data

    # Act
    response = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio
async def test_basic_error_response(datasource):
    """
    Test that the function correctly handles an error response with an 'error' attribute.
    """
    # Arrange: Mock a response object with an 'error' attribute
    class ErrorResponse:
        error = "Something went wrong"
    datasource.client.domain_dns_records.get.return_value = ErrorResponse()

    # Act
    response = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio
async def test_basic_error_dict(datasource):
    """
    Test that the function correctly handles an error response as a dict with 'error' key.
    """
    error_dict = {"error": {"code": "BadRequest", "message": "Invalid request"}}
    datasource.client.domain_dns_records.get.return_value = error_dict

    response = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio
async def test_basic_select_and_expand(datasource):
    """
    Test that select and expand parameters are correctly set and passed.
    """
    datasource.client.domain_dns_records.get.return_value = {"records": []}
    result = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(
        select=["id", "name"], expand=["related"]
    )
    # The actual query param logic is not externally visible, but we can check the call
    args, kwargs = datasource.client.domain_dns_records.get.call_args
    config = kwargs.get("request_configuration", args[0] if args else None)

# -------------------------------
# 2. Edge Test Cases
# -------------------------------

@pytest.mark.asyncio
async def test_edge_none_response(datasource):
    """
    Test that a None response from the API is handled as an error.
    """
    datasource.client.domain_dns_records.get.return_value = None
    result = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio
async def test_edge_exception_in_api_call(datasource):
    """
    Test that an exception raised during the API call is handled and returns a failed response.
    """
    datasource.client.domain_dns_records.get.side_effect = Exception("API failure!")
    result = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio
async def test_edge_headers_and_search_consistency_level(datasource):
    """
    Test that the ConsistencyLevel header is set when search is used, and that custom headers are preserved.
    """
    datasource.client.domain_dns_records.get.return_value = {"records": []}
    custom_headers = {"Authorization": "Bearer token"}
    await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(
        search="query", headers=custom_headers
    )
    args, kwargs = datasource.client.domain_dns_records.get.call_args
    config = kwargs.get("request_configuration", args[0] if args else None)

@pytest.mark.asyncio
async def test_edge_orderby_and_pagination(datasource):
    """
    Test orderby, top, skip parameters are passed and set.
    """
    datasource.client.domain_dns_records.get.return_value = {"records": []}
    await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(
        orderby="name", top=10, skip=5
    )
    args, kwargs = datasource.client.domain_dns_records.get.call_args
    config = kwargs.get("request_configuration", args[0] if args else None)

@pytest.mark.asyncio
async def test_edge_error_response_with_code_and_message(datasource):
    """
    Test that a response object with 'code' and 'message' attributes is handled as an error.
    """
    class CodeMessageError:
        code = "Forbidden"
        message = "Access denied"
    datasource.client.domain_dns_records.get.return_value = CodeMessageError()
    result = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio
async def test_edge_kwargs_are_accepted(datasource):
    """
    Test that extra kwargs do not cause errors and are ignored.
    """
    datasource.client.domain_dns_records.get.return_value = {"records": []}
    result = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(
        foo="bar", bar=123
    )

@pytest.mark.asyncio
async def test_edge_concurrent_calls(datasource):
    """
    Test concurrent execution of the async function.
    """
    # Each call should get a different result
    datasource.client.domain_dns_records.get.side_effect = [
        {"records": [1]}, {"records": [2]}, {"records": [3]}
    ]
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(),
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(),
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(),
    ]
    results = await asyncio.gather(*coros)

# -------------------------------
# 3. Large Scale Test Cases
# -------------------------------

@pytest.mark.asyncio
async def test_large_scale_many_concurrent_calls(datasource):
    """
    Test function performance under multiple concurrent calls (up to 50).
    """
    # All calls return the same dummy data
    datasource.client.domain_dns_records.get.return_value = {"records": ["dummy"]}
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(top=i)
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    for result in results:
        pass

@pytest.mark.asyncio
async def test_large_scale_varied_parameters(datasource):
    """
    Test that varied parameters in concurrent calls are handled correctly.
    """
    # Use a side effect to return different data for each call
    datasource.client.domain_dns_records.get.side_effect = [
        {"records": [i]} for i in range(10)
    ]
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(top=i)
        for i in range(10)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass

# -------------------------------
# 4. Throughput Test Cases
# -------------------------------

@pytest.mark.asyncio
async def test_domain_dns_records_domain_dns_record_list_domain_dns_record_throughput_small_load(datasource):
    """
    Throughput test: Run the function 10 times concurrently and ensure all succeed quickly.
    """
    datasource.client.domain_dns_records.get.return_value = {"records": ["ok"]}
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()
        for _ in range(10)
    ]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_domain_dns_records_domain_dns_record_list_domain_dns_record_throughput_medium_load(datasource):
    """
    Throughput test: Run the function 100 times concurrently and ensure all succeed.
    """
    datasource.client.domain_dns_records.get.return_value = {"records": ["medium"]}
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()
        for _ in range(100)
    ]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_domain_dns_records_domain_dns_record_list_domain_dns_record_throughput_high_volume(datasource):
    """
    Throughput test: Run the function 200 times concurrently with different parameters.
    """
    datasource.client.domain_dns_records.get.return_value = {"records": ["high"]}
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(top=i)
        for i in range(200)
    ]
    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.
#------------------------------------------------
import asyncio  # used to run async functions
import logging
# Helper to patch the 'users' attribute for client check
import types
from typing import Dict, List, Optional

import pytest  # used for our unit tests
from app.sources.external.microsoft.users_groups.users_groups import \
    UsersGroupsDataSource

# --- Minimal stubs and mocks for dependencies ---

# UsersGroupsResponse stub for test assertions
class UsersGroupsResponse:
    def __init__(self, success: bool, data=None, error: Optional[str] = None):
        self.success = success
        self.data = data
        self.error = error

# Simulate the domain_dns_records.get() async method
class MockDomainDnsRecords:
    def __init__(self, behavior=None):
        self.behavior = behavior or {}

    async def get(self, request_configuration=None):
        # Simulate different behaviors based on test
        if self.behavior.get("raise_exception"):
            raise self.behavior["raise_exception"]
        if self.behavior.get("return_error_dict"):
            return {"error": {"code": "BadRequest", "message": "Invalid filter"}}
        if self.behavior.get("return_error_attr"):
            class ErrorObj:
                error = "Some error"
            return ErrorObj()
        if self.behavior.get("return_none"):
            return None
        if self.behavior.get("return_custom"):
            return self.behavior["return_custom"]
        # Default: return a dummy list
        return [{"id": 1, "type": "A"}, {"id": 2, "type": "MX"}]

# Simulate the MSGraphServiceClient with domain_dns_records property
class MockMSGraphServiceClient:
    def __init__(self, domain_dns_records_behavior=None):
        self.domain_dns_records = MockDomainDnsRecords(domain_dns_records_behavior)

# Simulate the MSGraphClient that returns our mock service client
class MockMSGraphClient:
    def __init__(self, domain_dns_records_behavior=None):
        self._service_client = MockMSGraphServiceClient(domain_dns_records_behavior)
    def get_client(self):
        return self
    def get_ms_graph_service_client(self):
        return self._service_client

# 1. BASIC TEST CASES

@pytest.mark.asyncio
async def test_basic_returns_success_response(monkeypatch):
    # Test normal successful call
    # Patch 'users' attribute to bypass ValueError
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    datasource = UsersGroupsDataSource(MockMSGraphClient())
    resp = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio
async def test_basic_with_select_and_expand(monkeypatch):
    # Test select and expand parameters
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    datasource = UsersGroupsDataSource(MockMSGraphClient())
    resp = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(
        select=["id", "type"],
        expand=["relatedEntity"]
    )

@pytest.mark.asyncio
async def test_basic_with_headers_and_top(monkeypatch):
    # Test with custom headers and top parameter
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    datasource = UsersGroupsDataSource(MockMSGraphClient())
    resp = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(
        headers={"Authorization": "Bearer token"},
        top=1
    )

# 2. EDGE TEST CASES

@pytest.mark.asyncio
async def test_edge_returns_none(monkeypatch):
    # Simulate API returns None
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    client = MockMSGraphClient(domain_dns_records_behavior={"return_none": True})
    datasource = UsersGroupsDataSource(client)
    resp = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio
async def test_edge_returns_error_dict(monkeypatch):
    # Simulate API returns error dict
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    client = MockMSGraphClient(domain_dns_records_behavior={"return_error_dict": True})
    datasource = UsersGroupsDataSource(client)
    resp = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio
async def test_edge_returns_error_attr(monkeypatch):
    # Simulate API returns object with 'error' attribute
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    client = MockMSGraphClient(domain_dns_records_behavior={"return_error_attr": True})
    datasource = UsersGroupsDataSource(client)
    resp = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio
async def test_edge_raises_exception(monkeypatch):
    # Simulate API raises an exception (network error, etc.)
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    class DummyException(Exception):
        pass
    client = MockMSGraphClient(domain_dns_records_behavior={"raise_exception": DummyException("fail!")})
    datasource = UsersGroupsDataSource(client)
    resp = await datasource.domain_dns_records_domain_dns_record_list_domain_dns_record()

@pytest.mark.asyncio

async def test_edge_concurrent_execution(monkeypatch):
    # Test concurrent execution with different parameters
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    datasource = UsersGroupsDataSource(MockMSGraphClient())
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(top=1),
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(top=2),
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(select=["id"])
    ]
    results = await asyncio.gather(*coros)
    for resp in results:
        pass

# 3. LARGE SCALE TEST CASES

@pytest.mark.asyncio
async def test_large_scale_many_concurrent(monkeypatch):
    # Test 50 concurrent calls
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    datasource = UsersGroupsDataSource(MockMSGraphClient())
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(top=i)
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    for resp in results:
        pass

@pytest.mark.asyncio
async def test_large_scale_varied_parameters(monkeypatch):
    # Test with varied parameter combinations
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    datasource = UsersGroupsDataSource(MockMSGraphClient())
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(top=5, filter="type eq 'A'"),
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(expand=["related"]),
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(skip=10),
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(orderby="type desc"),
    ]
    results = await asyncio.gather(*coros)
    for resp in results:
        pass

# 4. THROUGHPUT TEST CASES

@pytest.mark.asyncio
async def test_domain_dns_records_domain_dns_record_list_domain_dns_record_throughput_small_load(monkeypatch):
    # Throughput test: 10 concurrent calls (small load)
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    datasource = UsersGroupsDataSource(MockMSGraphClient())
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(top=i)
        for i in range(10)
    ]
    results = await asyncio.gather(*coros)
    for resp in results:
        pass

@pytest.mark.asyncio
async def test_domain_dns_records_domain_dns_record_list_domain_dns_record_throughput_medium_load(monkeypatch):
    # Throughput test: 50 concurrent calls (medium load)
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    datasource = UsersGroupsDataSource(MockMSGraphClient())
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(top=i)
        for i in range(50)
    ]
    results = await asyncio.gather(*coros)
    for resp in results:
        pass

@pytest.mark.asyncio
async def test_domain_dns_records_domain_dns_record_list_domain_dns_record_throughput_high_load(monkeypatch):
    # Throughput test: 100 concurrent calls (high load, but <1000 as per instructions)
    monkeypatch.setattr(MockMSGraphServiceClient, "users", True, raising=False)
    datasource = UsersGroupsDataSource(MockMSGraphClient())
    coros = [
        datasource.domain_dns_records_domain_dns_record_list_domain_dns_record(top=i)
        for i in range(100)
    ]
    results = await asyncio.gather(*coros)
    for resp 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-UsersGroupsDataSource.domain_dns_records_domain_dns_record_list_domain_dns_record-mh6r0xlv and push.