perf: optimize route matching by caching static routes and adding a detection mechanism for custom routing classes

benchmark_routing.py

@@ -0,0 +1,94 @@
+import time
+from fastapi import APIRouter, FastAPI
+
+# Setup Router
+app = FastAPI()
+router = APIRouter()
+
+# 50 Static Routes
+for i in range(50):
+    @router.get(f"/static-{i}")
+    def static_route(i=i):
+        return {"route": f"static-{i}"}
+
+# 50 Dynamic Routes
+for i in range(50):
+    @router.get(f"/dynamic-{i}/{{item_id}}")
+    def dynamic_route(item_id: int, i=i):
+        return {"route": f"dynamic-{i}", "item_id": item_id}
+
+app.include_router(router)
+import anyio
+from contextlib import AsyncExitStack
+
+# Mock ASGI stack
+async def mock_receive():
+    return {"type": "http.request", "body": b"", "more_body": False}
+
+async def mock_send(message):
+    pass
+
+scope_static = {
+    "type": "http",
+    "method": "GET",
+    "path": "/static-49",
+    "headers": [],
+    "query_string": b"",
+    "client": ("127.0.0.1", 1234),
+    "server": ("127.0.0.1", 80),
+    "fastapi_middleware_astack": AsyncExitStack(),
+    "fastapi_inner_astack": AsyncExitStack(),
+    "fastapi_function_astack": AsyncExitStack(),
+}
+
+scope_dynamic = {
+    "type": "http",
+    "method": "GET",
+    "path": "/dynamic-49/12345",
+    "headers": [],
+    "query_string": b"",
+    "client": ("127.0.0.1", 1234),
+    "server": ("127.0.0.1", 80),
+    "fastapi_middleware_astack": AsyncExitStack(),
+    "fastapi_inner_astack": AsyncExitStack(),
+    "fastapi_function_astack": AsyncExitStack(),
+}
+
+async def run_benchmark():
+    # Warmup
+    for _ in range(100):
+        await app.router.app(dict(scope_static), mock_receive, mock_send)
+        await app.router.app(dict(scope_dynamic), mock_receive, mock_send)
+
+    print("Running Static Route Benchmark (10,000 iterations)...")
+
+    # 1. Caching Enabled (standard flow)
+    # Warmup cache
+    await app.router.app(dict(scope_static), mock_receive, mock_send)
+    start_time = time.perf_counter()
+    for _ in range(10000):
+        await app.router.app(dict(scope_static), mock_receive, mock_send)
+    cache_enabled_time = time.perf_counter() - start_time
+
+    # 2. Caching Disabled (clear cache before each request)
+    start_time = time.perf_counter()
+    for _ in range(10000):
+        app.router._route_cache.clear()
+        await app.router.app(dict(scope_static), mock_receive, mock_send)
+    cache_disabled_time = time.perf_counter() - start_time
+
+    print(f"Static Route - Cache Enabled:  {cache_enabled_time:.4f} seconds")
+    print(f"Static Route - Cache Disabled: {cache_disabled_time:.4f} seconds")
+    print(f"Speedup: {cache_disabled_time / cache_enabled_time:.2f}x")
+
+    print("\nRunning Dynamic Route Benchmark (10,000 iterations)...")
+    
+    # Caching has no effect on dynamic routes
+    start_time = time.perf_counter()
+    for _ in range(10000):
+        await app.router.app(dict(scope_dynamic), mock_receive, mock_send)
+    dynamic_time = time.perf_counter() - start_time
+    print(f"Dynamic Route time: {dynamic_time:.4f} seconds")
+
+if __name__ == "__main__":
+    anyio.run(run_benchmark)

fastapi/routing.py

@@ -2388,10 +2388,20 @@ class APIRouter(routing.Router):
             tuple[str, str | None, str],
             tuple[Match, BaseRoute, dict[str, Any], Any | None],
         ] = {}
+        self._custom_routing_detected: bool | None = None
 
     def _mark_routes_changed(self) -> None:
         self._routes_version += 1
         self._route_cache.clear()
+        self._custom_routing_detected = None
+
+    def _check_custom_routing(self) -> bool:
+        if type(self).matches is not APIRouter.matches or type(self).handle is not APIRouter.handle:
+            return True
+        for route in self.routes:
+            if isinstance(route, _IncludedRouter) and route.original_router._check_custom_routing():
+                return True
+        return False
 
     def _get_routes_version(self, seen: set[int] | None = None) -> int:
         if seen is None:
@@ -2540,36 +2550,61 @@ class APIRouter(routing.Router):
             await self.lifespan(scope, receive, send)
             return
 
-        # Check cache
+        if self._custom_routing_detected is None:
-        cache_key = (scope["type"], scope.get("method"), scope["path"])
+            self._custom_routing_detected = self._check_custom_routing()
-        cached = self._route_cache.get(cache_key)
+
-        if cached is not None:
+        if not self._custom_routing_detected:
-            match_type, leaf_route, child_scope, effective_context = cached
+            # Check cache
-            if match_type in (Match.FULL, Match.PARTIAL):
+            cache_key = (scope["type"], scope.get("method"), scope["path"])
-                resolved_scope = dict(child_scope)
+            cached = self._route_cache.get(cache_key)
-                if "path_params" in resolved_scope:
+            if cached is not None:
-                    resolved_scope["path_params"] = dict(resolved_scope["path_params"])
+                match_type, leaf_route, child_scope, effective_context = cached
-                scope.update(resolved_scope)
+                if match_type in (Match.FULL, Match.PARTIAL):
-                if effective_context is not None:
+                    resolved_scope = dict(child_scope)
-                    _get_fastapi_scope(scope)[_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY] = (
+                    if "path_params" in resolved_scope:
-                        effective_context
+                        resolved_scope["path_params"] = dict(resolved_scope["path_params"])
-                    )
+                    scope.update(resolved_scope)
-                    scope["route"] = effective_context.original_route
+                    if effective_context is not None:
+                        _get_fastapi_scope(scope)[_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY] = effective_context
+                        scope["route"] = effective_context.original_route
+                    else:
+                        scope["route"] = leaf_route
+                    await leaf_route.handle(scope, receive, send)
+                    return
+
+            partial_optimized: tuple[BaseRoute, Scope, BaseRoute, Any] | None = None
+            for route in self.routes:
+                if isinstance(route, _IncludedRouter):
+                    match, child_scope, leaf_route, effective_context = route._match(scope)
                 else:
-                    scope["route"] = leaf_route
+                    match, child_scope = route.matches(scope)
-                await leaf_route.handle(scope, receive, send)
+                    leaf_route = route
-                return
+                    effective_context = None
+
+                if match == Match.FULL:
+                    is_static = False
+                    if isinstance(leaf_route, APIRoute):
+                        is_static = len(leaf_route.dependant.path_params) == 0
+                    elif isinstance(leaf_route, routing.Route):
+                        is_static = not leaf_route.param_convertors
+
+                    if is_static:
+                        self._route_cache[cache_key] = (Match.FULL, leaf_route, child_scope, effective_context)
+
+                    scope.update(child_scope)
+                    if effective_context is not None:
+                        _get_fastapi_scope(scope)[_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY] = effective_context
+                        scope["route"] = effective_context.original_route
+                    else:
+                        scope["route"] = leaf_route
+                    await leaf_route.handle(scope, receive, send)
+                    return
 
-        partial: tuple[BaseRoute, Scope, BaseRoute, Any] | None = None
+                if match == Match.PARTIAL and partial_optimized is None:
-        for route in self.routes:
+                    partial_optimized = (route, child_scope, leaf_route, effective_context)
-            if isinstance(route, _IncludedRouter):
-                match, child_scope, leaf_route, effective_context = route._match(scope)
-            else:
-                match, child_scope = route.matches(scope)
-                leaf_route = route
-                effective_context = None
 
-            if match == Match.FULL:
+            if partial_optimized is not None:
+                _, child_scope, leaf_route, effective_context = partial_optimized
                 is_static = False
                 if isinstance(leaf_route, APIRoute):
                     is_static = len(leaf_route.dependant.path_params) == 0
@@ -2577,12 +2612,16 @@ class APIRouter(routing.Router):
                     is_static = not leaf_route.param_convertors
 
                 if is_static:
+<<<<<<< HEAD
                     self._route_cache[cache_key] = (
                         Match.FULL,
                         leaf_route,
                         child_scope,
                         effective_context,
                     )
+=======
+                    self._route_cache[cache_key] = (Match.PARTIAL, leaf_route, child_scope, effective_context)
+>>>>>>> fb6c05704 (perf: optimize route matching by caching static routes and adding a detection mechanism for custom routing classes)
 
                 scope.update(child_scope)
                 if effective_context is not None:
@@ -2595,9 +2634,19 @@ class APIRouter(routing.Router):
                 await leaf_route.handle(scope, receive, send)
                 return
 
-            if match == Match.PARTIAL and partial is None:
+        else:
-                partial = (route, child_scope, leaf_route, effective_context)
+            # Fallback to standard Starlette matching loop to preserve custom router matches/handle subclasses
+            partial: tuple[BaseRoute, Scope] | None = None
+            for route in self.routes:
+                match, child_scope = route.matches(scope)
+                if match == Match.FULL:
+                    scope.update(child_scope)
+                    await route.handle(scope, receive, send)
+                    return
+                if match == Match.PARTIAL and partial is None:
+                    partial = (route, child_scope)
 
+<<<<<<< HEAD
         if partial is not None:
             _, child_scope, leaf_route, effective_context = partial
             is_static = False
@@ -2624,6 +2673,13 @@ class APIRouter(routing.Router):
                 scope["route"] = leaf_route
             await leaf_route.handle(scope, receive, send)
             return
+=======
+            if partial is not None:
+                route, child_scope = partial
+                scope.update(child_scope)
+                await route.handle(scope, receive, send)
+                return
+>>>>>>> fb6c05704 (perf: optimize route matching by caching static routes and adding a detection mechanism for custom routing classes)
 
         route_path = get_route_path(scope)
         if scope["type"] == "http" and self.redirect_slashes and route_path != "/":