⚡️ Speed up method AsyncRawRequestsClient.get by 7%

[codeflash-ai]

📄 7% (0.07x) speedup for AsyncRawRequestsClient.get in src/deepgram/manage/v1/projects/requests/raw_client.py

⏱️ Runtime : 15.4 milliseconds → 14.5 milliseconds (best of 79 runs)

📝 Explanation and details

The optimized code achieves a 6% runtime speedup and 1.3% throughput improvement through two key optimizations:

1. Fast-path optimization in jsonable_encoder:

• Moved the most common type checks (str, int, float, None) to the very beginning of the function, before any other processing
• This eliminates unnecessary work for simple types that comprise the majority of encoding operations
• Line profiler shows the fast-path now handles 880 hits in just 0.49ms vs 3.12ms in the original, a 84% reduction in encoding time

2. Streamlined header and parameter processing in AsyncHttpClient.request:

• Replaced nested dictionary merging with flattened sequential updates using .update()
• Eliminated redundant isinstance checks and temporary variable creation
• Simplified conditional logic for file processing and parameter filtering
• Combined multiple dictionary operations into single passes

Why these optimizations work:

• Type checking order matters: In JSON encoding, primitive types (strings, numbers) are far more common than complex objects. Moving these checks first eliminates expensive isinstance calls for Pydantic models and dataclasses on simple data.
• Dictionary operations are costly: The original code created multiple intermediate dictionaries and performed redundant merging operations. The optimized version uses in-place updates and avoids unnecessary allocations.

Test case performance:
The optimizations are particularly effective for the test cases involving high-volume concurrent requests (test_get_throughput_high_volume with 100 concurrent requests), where the reduced per-operation overhead compounds significantly across many calls. The encoding improvements benefit all test cases since every HTTP request involves JSON encoding of headers and parameters.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio
# --- Function under test (EXACT COPY) ---
import typing
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from deepgram.manage.v1.projects.requests.raw_client import \
    AsyncRawRequestsClient


class ApiError(Exception):
    pass

class BadRequestError(Exception):
    pass

class GetProjectRequestV1Response:
    def __init__(self, value):
        self.value = value
    def __eq__(self, other):
        return isinstance(other, GetProjectRequestV1Response) and self.value == other.value

class AsyncHttpResponse:
    def __init__(self, response, data):
        self.response = response
        self.data = data
from deepgram.manage.v1.projects.requests.raw_client import \
    AsyncRawRequestsClient

# --- Helper mocks for the test suite ---

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

class DummyClientWrapper:
    def __init__(self, httpx_client, base_url="http://localhost"):
        self.httpx_client = httpx_client
        self._environment = DummyEnv(base_url)
    def get_environment(self):
        return self._environment

class DummyResponse:
    def __init__(self, status_code, 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

# --- Basic Test Cases ---

@pytest.mark.asyncio














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

import pytest  # used for our unit tests
from deepgram.manage.v1.projects.requests.raw_client import \
    AsyncRawRequestsClient


# Helper class to mock the client_wrapper and its httpx_client
class MockHttpxClient:
    def __init__(self, responses):
        self._responses = responses
        self._call_count = 0

    async def request(self, *args, **kwargs):
        resp = self._responses[self._call_count]
        self._call_count += 1
        return resp

class MockResponse:
    def __init__(self, status_code, json_data=None, headers=None, text=None):
        self.status_code = status_code
        self._json_data = json_data
        self.headers = headers or {}
        self.text = text or ""

    def json(self):
        if self._json_data is not None:
            return self._json_data
        raise Exception("JSONDecodeError")

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

class MockClientWrapper:
    def __init__(self, httpx_client, base_url="https://api.example.com"):
        self.httpx_client = httpx_client
        self._environment = MockEnvironment(base_url)

    def get_environment(self):
        return self._environment

# -------------------- Basic Test Cases --------------------

@pytest.mark.asyncio
async def test_get_successful_basic():
    """Test basic successful get request with status code 200."""
    # Setup
    response = MockResponse(200, json_data={"result": "ok"})
    httpx_client = MockHttpxClient([response])
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    # Execute
    result = await client.get("proj1", "req1")

@pytest.mark.asyncio
async def test_get_successful_status_299():
    """Test successful get request with status code 299 (upper bound of success)."""
    response = MockResponse(299, json_data={"result": "almost_ok"})
    httpx_client = MockHttpxClient([response])
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    result = await client.get("proj2", "req2")

@pytest.mark.asyncio
async def test_get_bad_request_raises():
    """Test get raises BadRequestError on status code 400."""
    response = MockResponse(400, json_data={"error": "bad request"})
    httpx_client = MockHttpxClient([response])
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    with pytest.raises(Exception) as excinfo:
        await client.get("proj3", "req3")

@pytest.mark.asyncio
async def test_get_api_error_non_400():
    """Test get raises ApiError on status code 404."""
    response = MockResponse(404, json_data={"error": "not found"})
    httpx_client = MockHttpxClient([response])
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    with pytest.raises(Exception) as excinfo:
        await client.get("proj4", "req4")

@pytest.mark.asyncio
async def test_get_api_error_json_decode():
    """Test get raises ApiError if .json() raises an exception."""
    response = MockResponse(500, json_data=None)
    # .json() will raise Exception
    httpx_client = MockHttpxClient([response])
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    with pytest.raises(Exception) as excinfo:
        await client.get("proj5", "req5")

@pytest.mark.asyncio
async def test_get_with_request_options():
    """Test get works with request_options argument."""
    response = MockResponse(200, json_data={"foo": "bar"})
    httpx_client = MockHttpxClient([response])
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    result = await client.get("proj6", "req6", request_options={"timeout_in_seconds": 1})

# -------------------- Edge Test Cases --------------------

@pytest.mark.asyncio
async def test_get_empty_project_and_request_id():
    """Test get with empty project_id and request_id."""
    response = MockResponse(200, json_data={"empty": True})
    httpx_client = MockHttpxClient([response])
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    result = await client.get("", "")

@pytest.mark.asyncio
async def test_get_none_request_options():
    """Test get with request_options explicitly set to None."""
    response = MockResponse(200, json_data={"none": "options"})
    httpx_client = MockHttpxClient([response])
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    result = await client.get("proj7", "req7", request_options=None)

@pytest.mark.asyncio
async def test_get_concurrent_successful():
    """Test multiple concurrent successful get requests."""
    responses = [MockResponse(200, json_data={"i": i}) for i in range(5)]
    httpx_client = MockHttpxClient(responses)
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    tasks = [client.get(f"proj{i}", f"req{i}") for i in range(5)]
    results = await asyncio.gather(*tasks)
    for idx, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_get_concurrent_mixed_results():
    """Test concurrent get requests with mixed status codes (200, 400, 404)."""
    responses = [
        MockResponse(200, json_data={"ok": 1}),
        MockResponse(400, json_data={"bad": 2}),
        MockResponse(404, json_data={"missing": 3}),
    ]
    httpx_client = MockHttpxClient(responses)
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    tasks = [
        client.get("projA", "reqA"),
        client.get("projB", "reqB"),
        client.get("projC", "reqC"),
    ]
    results = []
    for i, task in enumerate(tasks):
        if i == 0:
            # Should succeed
            res = await task
        else:
            # Should raise
            with pytest.raises(Exception) as excinfo:
                await task

@pytest.mark.asyncio

async def test_get_large_scale_concurrent():
    """Test get with many concurrent requests (scalability)."""
    n = 50  # Keep under 1000 for resource safety
    responses = [MockResponse(200, json_data={"idx": i}) for i in range(n)]
    httpx_client = MockHttpxClient(responses)
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    tasks = [client.get(f"proj{i}", f"req{i}") for i in range(n)]
    results = await asyncio.gather(*tasks)
    for i, result in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_get_large_scale_mixed_status():
    """Test get with many concurrent requests with mixed status codes."""
    n = 20
    responses = []
    for i in range(n):
        if i % 3 == 0:
            responses.append(MockResponse(400, json_data={"error": i}))
        elif i % 3 == 1:
            responses.append(MockResponse(404, json_data={"error": i}))
        else:
            responses.append(MockResponse(200, json_data={"ok": i}))
    httpx_client = MockHttpxClient(responses)
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    tasks = [client.get(f"proj{i}", f"req{i}") for i in range(n)]
    results = []
    for i, task in enumerate(tasks):
        if i % 3 == 2:
            # Should succeed
            res = await task
        else:
            # Should raise
            with pytest.raises(Exception) as excinfo:
                await task

# -------------------- Throughput Test Cases --------------------

@pytest.mark.asyncio


async def test_get_throughput_high_volume():
    """Throughput: Test get under high volume (100 concurrent requests)."""
    n = 100
    responses = [MockResponse(200, json_data={"hi": i}) for i in range(n)]
    httpx_client = MockHttpxClient(responses)
    client = AsyncRawRequestsClient(client_wrapper=MockClientWrapper(httpx_client))
    tasks = [client.get(f"proj{i}", f"req{i}") for i in range(n)]
    results = await asyncio.gather(*tasks)
    for i, result in enumerate(results):
        pass

@pytest.mark.asyncio

To edit these changes git checkout codeflash/optimize-AsyncRawRequestsClient.get-mh2sc6x7 and push.