⚡️ Speed up method AsyncRawModelsClient.list by 36%

[codeflash-ai]

📄 36% (0.36x) speedup for AsyncRawModelsClient.list in src/deepgram/manage/v1/models/raw_client.py

⏱️ Runtime : 15.9 milliseconds → 11.6 milliseconds (best of 121 runs)

📝 Explanation and details

The optimization introduces TypeAdapter caching in the Pydantic utilities, which significantly reduces object creation overhead during type parsing operations.

Key optimization:

• Added @lru_cache(maxsize=128) to TypeAdapter creation in Pydantic v2 path - instead of creating a new pydantic.TypeAdapter(type_) instance on every call, the function now caches adapters by type and reuses them

Why this improves performance:

• Eliminates redundant object instantiation - TypeAdapter objects are expensive to create but can be safely reused for the same type
• Reduces memory allocation pressure - fewer temporary objects created during parsing operations
• Optimizes hot path execution - the parse_obj_as function is called frequently (496 times in profiling), making caching highly effective

Performance impact from line profiler:

• 28% reduction in TypeAdapter creation time - from 23,023.7 ns to 2,243 ns per hit (90% faster per call)
• Overall parse_obj_as function time reduced by 28% - from 40.2ms to 28.9ms total execution time

The optimization is particularly effective for workloads with repeated type parsing operations using the same types, as shown in the test cases where the same ListModelsV1Response type is parsed multiple times. The caching provides consistent speedup across both single calls and concurrent execution scenarios, improving both runtime (36% faster) and throughput (6.1% increase to 60,863 operations/second).

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
# Function under test: AsyncRawModelsClient.list
# (copied exactly as provided above)
import typing
from json.decoder import JSONDecodeError

import pytest  # used for our unit tests
from deepgram.core.api_error import ApiError
from deepgram.core.client_wrapper import AsyncClientWrapper
from deepgram.core.http_response import AsyncHttpResponse
from deepgram.core.pydantic_utilities import parse_obj_as
from deepgram.core.request_options import RequestOptions
from deepgram.errors.bad_request_error import BadRequestError
from deepgram.manage.v1.models.raw_client import AsyncRawModelsClient
from deepgram.types.list_models_v1response import ListModelsV1Response


# --- Test Setup Utilities ---
class DummyResponse:
    """A dummy httpx.Response-like object for testing."""
    def __init__(self, status_code=200, json_data=None, headers=None, text='', should_raise_json_error=False):
        self.status_code = status_code
        self._json_data = json_data
        self.headers = headers or {}
        self.text = text
        self._should_raise_json_error = should_raise_json_error

    def json(self):
        if self._should_raise_json_error:
            raise JSONDecodeError("Expecting value", "doc", 0)
        return self._json_data

class DummyHttpxClient:
    """A dummy async httpx client for testing."""
    def __init__(self, response: DummyResponse):
        self._response = response
        self.requests = []

    async def request(self, *args, **kwargs):
        self.requests.append((args, kwargs))
        return self._response

class DummyEnvironment:
    def __init__(self, base="https://api.deepgram.com"):
        self.base = base

class DummyClientWrapper(AsyncClientWrapper):
    def __init__(self, httpx_client, environment):
        self.httpx_client = httpx_client
        self._environment = environment

    def get_environment(self):
        return self._environment

# --- Dummy ListModelsV1Response for parse_obj_as ---
class DummyListModelsV1Response:
    def __init__(self, models):
        self.models = models

# Monkeypatch parse_obj_as to return DummyListModelsV1Response for testing
def dummy_parse_obj_as(type_, object_):
    if type_ == ListModelsV1Response:
        return DummyListModelsV1Response(models=object_.get("models", []))
    return object_

# Monkeypatch AsyncHttpResponse for testing
class DummyAsyncHttpResponse:
    def __init__(self, response, data):
        self.response = response
        self.data = data

# --- Test Cases ---

@pytest.mark.asyncio




async def test_list_edge_400_bad_request(monkeypatch):
    """Edge: Should raise BadRequestError on 400 status code."""
    monkeypatch.setattr("deepgram.core.pydantic_utilities.parse_obj_as", dummy_parse_obj_as)
    models_data = {"error": "bad request"}
    dummy_response = DummyResponse(status_code=400, json_data=models_data)
    httpx_client = DummyHttpxClient(dummy_response)
    env = DummyEnvironment()
    wrapper = DummyClientWrapper(httpx_client, env)
    client = AsyncRawModelsClient(client_wrapper=wrapper)

    with pytest.raises(BadRequestError) as excinfo:
        await client.list()

@pytest.mark.asyncio
async def test_list_edge_non_2xx_non_400(monkeypatch):
    """Edge: Should raise ApiError on non-2xx, non-400 status code."""
    monkeypatch.setattr("deepgram.core.pydantic_utilities.parse_obj_as", dummy_parse_obj_as)
    models_data = {"error": "not found"}
    dummy_response = DummyResponse(status_code=404, json_data=models_data)
    httpx_client = DummyHttpxClient(dummy_response)
    env = DummyEnvironment()
    wrapper = DummyClientWrapper(httpx_client, env)
    client = AsyncRawModelsClient(client_wrapper=wrapper)

    with pytest.raises(ApiError) as excinfo:
        await client.list()

@pytest.mark.asyncio
async def test_list_edge_json_decode_error(monkeypatch):
    """Edge: Should raise ApiError on JSONDecodeError from response.json()."""
    monkeypatch.setattr("deepgram.core.pydantic_utilities.parse_obj_as", dummy_parse_obj_as)
    dummy_response = DummyResponse(status_code=500, should_raise_json_error=True, text="Internal Server Error")
    httpx_client = DummyHttpxClient(dummy_response)
    env = DummyEnvironment()
    wrapper = DummyClientWrapper(httpx_client, env)
    client = AsyncRawModelsClient(client_wrapper=wrapper)

    with pytest.raises(ApiError) as excinfo:
        await client.list()

@pytest.mark.asyncio
async def test_list_edge_none_response(monkeypatch):
    """Edge: Should handle None fields gracefully."""
    monkeypatch.setattr("deepgram.core.pydantic_utilities.parse_obj_as", dummy_parse_obj_as)
    monkeypatch.setattr("deepgram.core.http_response.AsyncHttpResponse", DummyAsyncHttpResponse)

    models_data = {"models": None}
    dummy_response = DummyResponse(status_code=200, json_data=models_data)
    httpx_client = DummyHttpxClient(dummy_response)
    env = DummyEnvironment()
    wrapper = DummyClientWrapper(httpx_client, env)
    client = AsyncRawModelsClient(client_wrapper=wrapper)

    result = await client.list()

@pytest.mark.asyncio
async def test_list_concurrent_execution(monkeypatch):
    """Edge: Multiple concurrent calls should all succeed independently."""
    monkeypatch.setattr("deepgram.core.pydantic_utilities.parse_obj_as", dummy_parse_obj_as)
    monkeypatch.setattr("deepgram.core.http_response.AsyncHttpResponse", DummyAsyncHttpResponse)

    models_data = {"models": [{"name": "model1"}]}
    dummy_response = DummyResponse(status_code=200, json_data=models_data)
    httpx_client = DummyHttpxClient(dummy_response)
    env = DummyEnvironment()
    wrapper = DummyClientWrapper(httpx_client, env)
    client = AsyncRawModelsClient(client_wrapper=wrapper)

    # Run 5 concurrent calls
    results = await asyncio.gather(*(client.list() for _ in range(5)))
    for result in results:
        pass

# --- Large Scale Tests ---

@pytest.mark.asyncio
async def test_list_large_scale_many_concurrent(monkeypatch):
    """Large Scale: 100 concurrent calls should all succeed independently."""
    monkeypatch.setattr("deepgram.core.pydantic_utilities.parse_obj_as", dummy_parse_obj_as)
    monkeypatch.setattr("deepgram.core.http_response.AsyncHttpResponse", DummyAsyncHttpResponse)

    models_data = {"models": [{"name": "model1"}]}
    dummy_response = DummyResponse(status_code=200, json_data=models_data)
    httpx_client = DummyHttpxClient(dummy_response)
    env = DummyEnvironment()
    wrapper = DummyClientWrapper(httpx_client, env)
    client = AsyncRawModelsClient(client_wrapper=wrapper)

    results = await asyncio.gather(*(client.list() for _ in range(100)))
    for result in results:
        pass

@pytest.mark.asyncio
async def test_list_large_scale_varied_inputs(monkeypatch):
    """Large Scale: Concurrent calls with varied inputs."""
    monkeypatch.setattr("deepgram.core.pydantic_utilities.parse_obj_as", dummy_parse_obj_as)
    monkeypatch.setattr("deepgram.core.http_response.AsyncHttpResponse", DummyAsyncHttpResponse)

    dummy_response_true = DummyResponse(status_code=200, json_data={"models": [{"name": "model1", "outdated": True}]})
    dummy_response_false = DummyResponse(status_code=200, json_data={"models": [{"name": "model2", "outdated": False}]})

    # Simulate two different httpx clients for two different inputs
    httpx_client_true = DummyHttpxClient(dummy_response_true)
    httpx_client_false = DummyHttpxClient(dummy_response_false)
    env = DummyEnvironment()
    wrapper_true = DummyClientWrapper(httpx_client_true, env)
    wrapper_false = DummyClientWrapper(httpx_client_false, env)
    client_true = AsyncRawModelsClient(client_wrapper=wrapper_true)
    client_false = AsyncRawModelsClient(client_wrapper=wrapper_false)

    results = await asyncio.gather(
        client_true.list(include_outdated=True),
        client_false.list(include_outdated=False),
    )

# --- Throughput Tests ---

@pytest.mark.asyncio



#------------------------------------------------
import asyncio  # used to run async functions
# Function to test (AsyncRawModelsClient.list) copied exactly as provided
import json
import typing
from json.decoder import JSONDecodeError
from unittest.mock import AsyncMock, MagicMock, patch

import pytest  # used for our unit tests
from deepgram.manage.v1.models.raw_client import AsyncRawModelsClient


# Simulate minimal dependencies for the test
class DummyResponse:
    def __init__(self, status_code=200, json_data=None, headers=None, text=""):
        self.status_code = status_code
        self._json_data = json_data
        self.headers = headers or {}
        self.text = text

    def json(self):
        if isinstance(self._json_data, Exception):
            raise self._json_data
        return self._json_data

class DummyEnv:
    def __init__(self, base):
        self.base = base

class DummyClientWrapper:
    def __init__(self, response: DummyResponse):
        self.httpx_client = AsyncMock()
        self.httpx_client.request = AsyncMock(return_value=response)
        self._env = DummyEnv("https://example.com")
    def get_environment(self):
        return self._env

# Minimal AsyncHttpResponse and ListModelsV1Response for test
class AsyncHttpResponse(typing.Generic[typing.TypeVar('T')]):
    def __init__(self, response, data):
        self.response = response
        self.data = data

class ListModelsV1Response:
    def __init__(self, models):
        self.models = models
    def __eq__(self, other):
        return isinstance(other, ListModelsV1Response) and self.models == other.models

# Simulate BadRequestError and ApiError
class BadRequestError(Exception):
    def __init__(self, headers, body):
        super().__init__("BadRequestError")
        self.headers = headers
        self.body = body

class ApiError(Exception):
    def __init__(self, status_code, headers, body):
        super().__init__("ApiError")
        self.status_code = status_code
        self.headers = headers
        self.body = body
from deepgram.manage.v1.models.raw_client import AsyncRawModelsClient

# ------------------ TESTS BEGIN HERE ------------------

# 1. Basic Test Cases

@pytest.mark.asyncio

To edit these changes git checkout codeflash/optimize-AsyncRawModelsClient.list-mh2sqzh8 and push.