⚡️ Speed up method HTTPEndpoint.method_not_allowed by 7%

[codeflash-ai]

📄 7% (0.07x) speedup for HTTPEndpoint.method_not_allowed in starlette/endpoints.py

⏱️ Runtime : 3.67 milliseconds → 3.42 milliseconds (best of 164 runs)

📝 Explanation and details

The optimization achieves a 7% runtime improvement by precomputing expensive string operations that were previously executed on every method_not_allowed call.

What changed:

• Precomputed string joining: The ", ".join(self._allowed_methods) operation is moved from the hot path in method_not_allowed to the initialization phase, stored as _allowed_methods_str
• Precomputed headers dictionary: The headers dict {"Allow": ...} is created once during __init__ as _method_not_allowed_headers instead of being recreated on each method call

Why this is faster:

• Eliminates repeated string operations: String joining is computationally expensive and was happening on every request. The line profiler shows this operation took 725ms total (5.5% of runtime) across 1748 calls in the original code
• Reduces object allocations: Dictionary creation for headers happened on every call, now it's done once during initialization
• CPU cache efficiency: Reusing the same precomputed objects improves memory access patterns

Performance impact by test case:

• High-throughput scenarios (500+ concurrent requests) benefit most from eliminating repeated computations
• Mixed workloads with both exception and response paths see consistent improvements since both code paths use the precomputed headers
• Edge cases with custom method combinations still benefit from the one-time string joining optimization

The 2.5% throughput improvement (279,680 → 286,672 ops/sec) demonstrates better request processing capacity, making this optimization particularly valuable for high-traffic web applications where method_not_allowed responses are frequent.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 1824 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
from typing import Any, Generator

import pytest  # used for our unit tests
from starlette.endpoints import HTTPEndpoint
# function to test
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import PlainTextResponse, Response
from starlette.types import Receive, Scope, Send

# Helper functions and fixtures

class DummyReceive:
    async def __call__(self):
        return {}

class DummySend:
    async def __call__(self, message):
        pass

class DummyRequest(Request):
    # Minimal Request implementation for testing
    def __init__(self, scope):
        self.scope = scope

# --- Basic Test Cases ---

@pytest.mark.asyncio
async def test_method_not_allowed_plain_asgi_returns_plaintext_response():
    """
    Basic: Test that method_not_allowed returns a PlainTextResponse with correct status and headers
    when 'app' is not in scope (plain ASGI app).
    """
    # Setup: scope without 'app', allowed_methods = []
    scope = {"type": "http"}
    endpoint = HTTPEndpoint(scope, DummyReceive(), DummySend())
    request = DummyRequest(scope)
    response = await endpoint.method_not_allowed(request)

@pytest.mark.asyncio
async def test_method_not_allowed_plain_asgi_with_allowed_methods():
    """
    Basic: Test that Allow header is correct when allowed methods exist.
    """
    class CustomEndpoint(HTTPEndpoint):
        async def get(self, request): pass
        async def post(self, request): pass

    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, DummyReceive(), DummySend())
    request = DummyRequest(scope)
    response = await endpoint.method_not_allowed(request)

@pytest.mark.asyncio
async def test_method_not_allowed_async_await_behavior():
    """
    Basic: Test that method_not_allowed is a coroutine and can be awaited.
    """
    scope = {"type": "http"}
    endpoint = HTTPEndpoint(scope, DummyReceive(), DummySend())
    request = DummyRequest(scope)
    codeflash_output = endpoint.method_not_allowed(request); coro = codeflash_output
    response = await coro

# --- Edge Test Cases ---

@pytest.mark.asyncio
async def test_method_not_allowed_raises_http_exception_when_app_in_scope():
    """
    Edge: Test that method_not_allowed raises HTTPException when 'app' is in scope.
    """
    scope = {"type": "http", "app": object()}
    endpoint = HTTPEndpoint(scope, DummyReceive(), DummySend())
    request = DummyRequest(scope)
    with pytest.raises(HTTPException) as exc_info:
        await endpoint.method_not_allowed(request)
    exc = exc_info.value

@pytest.mark.asyncio
async def test_method_not_allowed_raises_http_exception_with_allowed_methods():
    """
    Edge: If allowed methods are present, HTTPException headers should reflect them.
    """
    class CustomEndpoint(HTTPEndpoint):
        async def put(self, request): pass
        async def patch(self, request): pass

    scope = {"type": "http", "app": object()}
    endpoint = CustomEndpoint(scope, DummyReceive(), DummySend())
    request = DummyRequest(scope)
    with pytest.raises(HTTPException) as exc_info:
        await endpoint.method_not_allowed(request)
    exc = exc_info.value

@pytest.mark.asyncio
async def test_method_not_allowed_concurrent_exceptions_and_responses():
    """
    Edge: Test concurrent calls with mixed scopes: some should raise, some should return response.
    """
    class CustomEndpoint(HTTPEndpoint):
        async def get(self, request): pass

    # One with 'app' in scope, one without
    scope1 = {"type": "http", "app": object()}
    scope2 = {"type": "http"}
    endpoint1 = CustomEndpoint(scope1, DummyReceive(), DummySend())
    endpoint2 = CustomEndpoint(scope2, DummyReceive(), DummySend())
    request1 = DummyRequest(scope1)
    request2 = DummyRequest(scope2)

    async def call1():
        with pytest.raises(HTTPException) as exc_info:
            await endpoint1.method_not_allowed(request1)
        exc = exc_info.value

    async def call2():
        response = await endpoint2.method_not_allowed(request2)

    await asyncio.gather(call1(), call2())

@pytest.mark.asyncio
async def test_method_not_allowed_edge_empty_allowed_methods():
    """
    Edge: Test that Allow header is empty if no methods are implemented.
    """
    scope = {"type": "http"}
    endpoint = HTTPEndpoint(scope, DummyReceive(), DummySend())
    request = DummyRequest(scope)
    response = await endpoint.method_not_allowed(request)

# --- Large Scale Test Cases ---

@pytest.mark.asyncio
async def test_method_not_allowed_many_concurrent_plain_asgi():
    """
    Large Scale: Test many concurrent calls, all plain ASGI, all should return PlainTextResponse.
    """
    class CustomEndpoint(HTTPEndpoint):
        async def get(self, request): pass
        async def post(self, request): pass

    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, DummyReceive(), DummySend())
    requests = [DummyRequest(scope) for _ in range(100)]
    coros = [endpoint.method_not_allowed(req) for req in requests]
    responses = await asyncio.gather(*coros)
    for response in responses:
        pass

@pytest.mark.asyncio
async def test_method_not_allowed_many_concurrent_with_app_and_without():
    """
    Large Scale: Mix of scopes with and without 'app', test concurrent exceptions and responses.
    """
    class CustomEndpoint(HTTPEndpoint):
        async def delete(self, request): pass

    scopes = []
    endpoints = []
    requests = []
    for i in range(50):
        if i % 2 == 0:
            scope = {"type": "http", "app": object()}
        else:
            scope = {"type": "http"}
        scopes.append(scope)
        endpoints.append(CustomEndpoint(scope, DummyReceive(), DummySend()))
        requests.append(DummyRequest(scope))

    async def call(endpoint, request, expect_exception):
        if expect_exception:
            with pytest.raises(HTTPException) as exc_info:
                await endpoint.method_not_allowed(request)
            exc = exc_info.value
        else:
            response = await endpoint.method_not_allowed(request)

    coros = [
        call(endpoints[i], requests[i], expect_exception=(i % 2 == 0))
        for i in range(50)
    ]
    await asyncio.gather(*coros)

# --- Throughput Test Cases ---

@pytest.mark.asyncio
async def test_method_not_allowed_throughput_small_load():
    """
    Throughput: Test small load (10 concurrent calls).
    """
    class CustomEndpoint(HTTPEndpoint):
        async def get(self, request): pass

    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, DummyReceive(), DummySend())
    requests = [DummyRequest(scope) for _ in range(10)]
    coros = [endpoint.method_not_allowed(req) for req in requests]
    responses = await asyncio.gather(*coros)
    for response in responses:
        pass

@pytest.mark.asyncio
async def test_method_not_allowed_throughput_medium_load():
    """
    Throughput: Test medium load (100 concurrent calls).
    """
    class CustomEndpoint(HTTPEndpoint):
        async def post(self, request): pass

    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, DummyReceive(), DummySend())
    requests = [DummyRequest(scope) for _ in range(100)]
    coros = [endpoint.method_not_allowed(req) for req in requests]
    responses = await asyncio.gather(*coros)
    for response in responses:
        pass

@pytest.mark.asyncio
async def test_method_not_allowed_throughput_high_volume():
    """
    Throughput: Test high volume (500 concurrent calls).
    """
    class CustomEndpoint(HTTPEndpoint):
        async def put(self, request): pass
        async def patch(self, request): pass

    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, DummyReceive(), DummySend())
    requests = [DummyRequest(scope) for _ in range(500)]
    coros = [endpoint.method_not_allowed(req) for req in requests]
    responses = await asyncio.gather(*coros)
    for response in responses:
        pass
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
from __future__ import annotations

import asyncio  # used to run async functions
from collections.abc import Generator
from typing import Any

import pytest  # used for our unit tests
from starlette.endpoints import HTTPEndpoint
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import PlainTextResponse, Response
from starlette.types import Receive, Scope, Send


# Helper: Dummy ASGI receive/send functions
async def dummy_receive():
    return {}

async def dummy_send(message):
    pass

# Helper: Minimal Starlette Request mock
class MockRequest:
    def __init__(self, method="GET"):
        self.method = method

# Helper: Custom endpoint with only GET/POST allowed
class CustomEndpoint(HTTPEndpoint):
    async def get(self, request):
        pass
    async def post(self, request):
        pass

# Helper: Custom endpoint with no allowed methods
class NoMethodEndpoint(HTTPEndpoint):
    pass

# Helper: Custom endpoint with all allowed methods
class AllMethodEndpoint(HTTPEndpoint):
    async def get(self, request): pass
    async def head(self, request): pass
    async def post(self, request): pass
    async def put(self, request): pass
    async def patch(self, request): pass
    async def delete(self, request): pass
    async def options(self, request): pass

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

@pytest.mark.asyncio
async def test_method_not_allowed_returns_plaintext_response_without_app():
    """Test that method_not_allowed returns a 405 PlainTextResponse when no 'app' in scope."""
    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, dummy_receive, dummy_send)
    request = MockRequest("PUT")
    response = await endpoint.method_not_allowed(request)

@pytest.mark.asyncio
async def test_method_not_allowed_raises_http_exception_with_app():
    """Test that method_not_allowed raises HTTPException when 'app' in scope."""
    scope = {"type": "http", "app": object()}
    endpoint = CustomEndpoint(scope, dummy_receive, dummy_send)
    request = MockRequest("DELETE")
    with pytest.raises(HTTPException) as excinfo:
        await endpoint.method_not_allowed(request)
    exc = excinfo.value

@pytest.mark.asyncio
async def test_method_not_allowed_no_allowed_methods():
    """Test with endpoint that has no allowed methods."""
    scope = {"type": "http"}
    endpoint = NoMethodEndpoint(scope, dummy_receive, dummy_send)
    request = MockRequest("GET")
    response = await endpoint.method_not_allowed(request)

@pytest.mark.asyncio
async def test_method_not_allowed_all_methods_allowed():
    """Test with endpoint that has all methods allowed."""
    scope = {"type": "http"}
    endpoint = AllMethodEndpoint(scope, dummy_receive, dummy_send)
    request = MockRequest("PATCH")
    response = await endpoint.method_not_allowed(request)

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

@pytest.mark.asyncio
async def test_method_not_allowed_concurrent_without_app():
    """Test concurrent execution of method_not_allowed without 'app' in scope."""
    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, dummy_receive, dummy_send)
    requests = [MockRequest("PUT") for _ in range(10)]
    responses = await asyncio.gather(*(endpoint.method_not_allowed(req) for req in requests))
    for response in responses:
        pass

@pytest.mark.asyncio
async def test_method_not_allowed_concurrent_with_app():
    """Test concurrent execution of method_not_allowed with 'app' in scope."""
    scope = {"type": "http", "app": object()}
    endpoint = CustomEndpoint(scope, dummy_receive, dummy_send)
    requests = [MockRequest("DELETE") for _ in range(5)]
    # All should raise HTTPException
    async def call():
        with pytest.raises(HTTPException) as excinfo:
            await endpoint.method_not_allowed(MockRequest("DELETE"))
        exc = excinfo.value
    await asyncio.gather(*(call() for _ in range(5)))

@pytest.mark.asyncio
async def test_method_not_allowed_custom_methods():
    """Test endpoint with custom allowed methods."""
    class CustomMethodsEndpoint(HTTPEndpoint):
        async def get(self, request): pass
        async def patch(self, request): pass
    scope = {"type": "http"}
    endpoint = CustomMethodsEndpoint(scope, dummy_receive, dummy_send)
    request = MockRequest("OPTIONS")
    response = await endpoint.method_not_allowed(request)

@pytest.mark.asyncio
async def test_method_not_allowed_invalid_scope_type():
    """Test that constructor asserts on invalid scope type."""
    invalid_scope = {"type": "websocket"}
    with pytest.raises(AssertionError):
        CustomEndpoint(invalid_scope, dummy_receive, dummy_send)

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

@pytest.mark.asyncio
async def test_method_not_allowed_many_concurrent_calls():
    """Test large number of concurrent calls to method_not_allowed."""
    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, dummy_receive, dummy_send)
    requests = [MockRequest("PUT") for _ in range(100)]
    responses = await asyncio.gather(*(endpoint.method_not_allowed(req) for req in requests))
    for response in responses:
        pass

@pytest.mark.asyncio
async def test_method_not_allowed_many_concurrent_calls_with_app():
    """Test large number of concurrent calls to method_not_allowed with 'app' in scope."""
    scope = {"type": "http", "app": object()}
    endpoint = CustomEndpoint(scope, dummy_receive, dummy_send)
    async def call():
        with pytest.raises(HTTPException) as excinfo:
            await endpoint.method_not_allowed(MockRequest("DELETE"))
        exc = excinfo.value
    await asyncio.gather(*(call() for _ in range(50)))

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

@pytest.mark.asyncio
async def test_method_not_allowed_throughput_small_load():
    """Throughput test: small load (10 requests) without 'app' in scope."""
    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, dummy_receive, dummy_send)
    requests = [MockRequest("PUT") for _ in range(10)]
    responses = await asyncio.gather(*(endpoint.method_not_allowed(req) for req in requests))

@pytest.mark.asyncio
async def test_method_not_allowed_throughput_medium_load():
    """Throughput test: medium load (100 requests) without 'app' in scope."""
    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, dummy_receive, dummy_send)
    requests = [MockRequest("PUT") for _ in range(100)]
    responses = await asyncio.gather(*(endpoint.method_not_allowed(req) for req in requests))

@pytest.mark.asyncio
async def test_method_not_allowed_throughput_high_load():
    """Throughput test: high load (500 requests) without 'app' in scope."""
    scope = {"type": "http"}
    endpoint = CustomEndpoint(scope, dummy_receive, dummy_send)
    requests = [MockRequest("PUT") for _ in range(500)]
    responses = await asyncio.gather(*(endpoint.method_not_allowed(req) for req in requests))

@pytest.mark.asyncio
async def test_method_not_allowed_throughput_high_load_with_app():
    """Throughput test: high load (200 requests) with 'app' in scope, all should raise HTTPException."""
    scope = {"type": "http", "app": object()}
    endpoint = CustomEndpoint(scope, dummy_receive, dummy_send)
    async def call():
        with pytest.raises(HTTPException) as excinfo:
            await endpoint.method_not_allowed(MockRequest("DELETE"))
        exc = excinfo.value
    await asyncio.gather(*(call() for _ in range(200)))
# 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-HTTPEndpoint.method_not_allowed-mhbj2l5w and push.