Merge b11e2f64bd into 5b28a04d55

2025-12-09 11:14:40 +00:00 · 2025-12-09 11:14:40 +00:00 · 965e0083d5
parent 5b28a04d55 b11e2f64bd
commit 965e0083d5
5 changed files with 243 additions and 4 deletions
--- a/docs/en/docs/how-to/custom-request-and-route.md
+++ b/docs/en/docs/how-to/custom-request-and-route.md
@ -107,3 +107,35 @@ You can also set the `route_class` parameter of an `APIRouter`:
 In this example, the *path operations* under the `router` will use the custom `TimedRoute` class, and will have an extra `X-Response-Time` header in the response with the time it took to generate the response:

 {* ../../docs_src/custom_request_and_route/tutorial003.py hl[13:20] *}
+
+## Custom `APIRouter` class in a router
+
+You can also set the `router_class` parameter of an `APIRouter`:
+
+{* ../../docs_src/custom_api_router/tutorial001.py hl[100:102] *}
+
+#### 🚀 Custom FastAPI Router with Timed Responses
+
+This example enhances FastAPI with structured routing and response timing, making APIs more organized and observable.
+
+##### ✨ Features
+
+- **`TimedRoute`**: Measures request duration and adds `X-Response-Time` to response headers.
+- **`AppRouter`**: A custom router that:
+  - Supports **nested routers** with automatic hierarchical route naming.
+  - Includes a **built-in `/healthz` endpoint** for every router.
+  - Ensures **clean API structure** with logical parent-child relationships.
+##### 📌 API Structure
+- **`/healthz`**: Health check endpoint for the main router. it path name is `Global.health-check`.
+- **`/model/create`**: Model creation endpoint for the model router with path name `Model.create`.
+- **`/model/{model_id}/item/create`**: Item creation endpoint for the item router and its child router of model
+  router with path name `Model.Item.create`.
+##### 🔥 Benefits
+
+- **Clear & maintainable API design** with structured route naming.
+- **Built-in health checks** for easier observability.
+- **Performance monitoring** with request duration logging.
+
+This setup is **ideal for scalable FastAPI projects**, ensuring better organization and easier debugging.
+
+{* ../../docs_src/custom_api_router/tutorial001.py hl[100:149,21:27] *}
--- a/docs_src/custom_api_router/tutorial001.py
+++ b/docs_src/custom_api_router/tutorial001.py
@ -0,0 +1,149 @@
+import time
+from typing import Any, Awaitable, Callable, List, Optional, Set, Union
+
+from fastapi import APIRouter, FastAPI, Request, Response
+from fastapi.responses import JSONResponse
+from fastapi.routing import APIRoute
+
+
+async def health_check(request: Request):
+    """
+    Health check endpoint
+    """
+    return Response(content="OK", status_code=200)
+
+
+class TimedRoute(APIRoute):
+    def get_route_handler(self) -> Callable:
+        original_route_handler = super().get_route_handler()
+
+        async def custom_route_handler(request: Request) -> Response:
+            before = time.time()
+            response: Response = await original_route_handler(request)
+            duration = time.time() - before
+            response.headers["X-Response-Time"] = str(duration)
+            print(f"{self.name} route duration: {duration}")
+            print(f"{self.name} route response: {response}")
+            print(f"{self.name} route response headers: {response.headers}")
+            return response
+
+        return custom_route_handler
+
+
+class AppRouter(APIRouter):
+    def __init__(self, prefix="", name="Global", tags: list = None, **kwargs):
+        self.name = name
+        tags = tags or []
+        tags.insert(0, name)
+        super().__init__(prefix=prefix, tags=tags, **kwargs)
+        self._parent: Optional[AppRouter] = None
+        self._add_health_check()
+
+    @property
+    def request_name_prefix(self):
+        return (
+            f"{self._parent.request_name_prefix}.{self.name}"
+            if self._parent
+            else self.name
+        )
+
+    def _add_health_check(self):
+        """
+        Adding default health check route for all new routers
+        """
+        self.add_api_route(
+            "/healthz", endpoint=health_check, methods=["GET"], name="health-check"
+        )
+
+    def include_router(self, router: "AppRouter", **kwargs):
+        """
+        Include another router into this router.
+        """
+        router._parent = self
+        super().include_router(router, **kwargs)
+
+    def add_api_route(
+        self,
+        path: str,
+        endpoint: Callable[..., Any],
+        methods: Union[Set[str], List[str]],  # noqa
+        name: str,
+        **kwargs,
+    ):
+        name = f"{self.request_name_prefix}.{name}"
+        return super().add_api_route(
+            path,
+            endpoint,
+            methods=methods,
+            name=name,
+            **kwargs,
+        )
+
+    def add_route(
+        self,
+        path: str,
+        endpoint: Callable[[Request], Union[Awaitable[Response], Response]],
+        methods: Union[List[str], None] = None,
+        name: Union[str, None] = None,
+        include_in_schema: bool = True,
+    ) -> None:
+        name = f"{self.request_name_prefix}.{name}"
+        return super().add_route(
+            path,
+            endpoint,
+            methods=methods,
+            name=name,
+            include_in_schema=include_in_schema,
+        )
+
+
+app = FastAPI(route_class=TimedRoute, router_class=AppRouter)
+model = AppRouter(prefix="/model", name="Model", route_class=TimedRoute)
+item = AppRouter(prefix="/{model_id}/item", name="Item", route_class=TimedRoute)
+
+
+async def create_model(request: Request):
+    """
+    Create a model
+    """
+    print("Model created")
+    route: TimedRoute = request.scope["route"]
+    router: AppRouter = request.scope["router"]
+    return JSONResponse(
+        {
+            "route_class": route.__class__.__name__,
+            "route_name": route.name,
+            "router_class": router.__class__.__name__,
+        },
+        status_code=200,
+    )
+
+
+model.add_api_route(
+    path="/create", endpoint=create_model, methods=["POST"], name="create-model"
+)
+
+
+async def create_item(request: Request):
+    """
+    Create an item
+    """
+    print("Item created")
+    route: TimedRoute = request.scope["route"]
+    router: AppRouter = request.scope["router"]
+    return JSONResponse(
+        {
+            "route_class": route.__class__.__name__,
+            "route_name": route.name,
+            "router_class": router.__class__.__name__,
+        },
+        status_code=200,
+    )
+
+
+item.add_api_route(
+    path="/create", endpoint=create_item, methods=["POST"], name="create-item"
+)
+
+model.include_router(item)
+app.include_router(model)
--- a/fastapi/applications.py
+++ b/fastapi/applications.py
@ -819,6 +819,28 @@ class FastAPI(Starlette):
                """
            ),
        ] = True,
+        route_class: Annotated[
+            Type[routing.APIRoute],
+            Doc(
+                """
+                Custom route (*path operation*) class to be used by this router.
+
+                Read more about it in the
+                [FastAPI docs for Custom Request and APIRoute class](https://fastapi.tiangolo.com/how-to/custom-request-and-route/#custom-apiroute-class-in-a-router).
+                """
+            ),
+        ] = routing.APIRoute,
+        router_class: Annotated[
+            Type[routing.APIRouter],
+            Doc(
+                """
+                Custom router class to be used by this application.
+
+                Read more about it in the
+                [FastAPI docs for Custom Request and APIRouter class](https://fastapi.tiangolo.com/how-to/custom-request-and-route/#custom-apirouter-class-in-a-router).
+                """
+            ),
+        ] = routing.APIRouter,
        openapi_external_docs: Annotated[
            Optional[Dict[str, Any]],
            Doc(
@ -929,7 +951,7 @@ class FastAPI(Starlette):
                [FastAPI docs for OpenAPI Webhooks](https://fastapi.tiangolo.com/advanced/openapi-webhooks/).
                """
            ),
-        ] = webhooks or routing.APIRouter()
+        ] = webhooks or router_class(route_class=route_class)
        self.root_path = root_path or openapi_prefix
        self.state: Annotated[
            State,
@ -965,7 +987,7 @@ class FastAPI(Starlette):
                """
            ),
        ] = {}
-        self.router: routing.APIRouter = routing.APIRouter(
+        self.router: routing.APIRouter = router_class(
            routes=routes,
            redirect_slashes=redirect_slashes,
            dependency_overrides_provider=self,
@ -979,6 +1001,7 @@ class FastAPI(Starlette):
            include_in_schema=include_in_schema,
            responses=responses,
            generate_unique_id_function=generate_unique_id_function,
+            route_class=route_class,
        )
        self.exception_handlers: Dict[
            Any, Callable[[Request, Any], Union[Response, Awaitable[Response]]]
--- a/fastapi/routing.py
+++ b/fastapi/routing.py
@ -1045,7 +1045,7 @@ class APIRouter(routing.Router):
        generate_unique_id_function: Union[
            Callable[[APIRoute], str], DefaultPlaceholder
        ] = Default(generate_unique_id),
-    ) -> None:
+    ) -> "APIRoute":
        route_class = route_class_override or self.route_class
        responses = responses or {}
        combined_responses = {**self.responses, **responses}
@ -1093,6 +1093,7 @@ class APIRouter(routing.Router):
            generate_unique_id_function=current_generate_unique_id,
        )
        self.routes.append(route)
+        return route

    def api_route(
        self,
@ -1163,7 +1164,7 @@ class APIRouter(routing.Router):
        name: Optional[str] = None,
        *,
        dependencies: Optional[Sequence[params.Depends]] = None,
-    ) -> None:
+    ) -> "APIWebSocketRoute":
        current_dependencies = self.dependencies.copy()
        if dependencies:
            current_dependencies.extend(dependencies)
@ -1176,6 +1177,7 @@ class APIRouter(routing.Router):
            dependency_overrides_provider=self.dependency_overrides_provider,
        )
        self.routes.append(route)
+        return route

    def websocket(
        self,
--- a/tests/test_tutorial/test_custom_request_and_route/test_tutorial004.py
+++ b/tests/test_tutorial/test_custom_request_and_route/test_tutorial004.py
@ -0,0 +1,33 @@
+from fastapi.testclient import TestClient
+
+from docs_src.custom_api_router.tutorial001 import app
+
+client = TestClient(app)
+
+
+def test_get_timed():
+    response = client.get("/healthz")
+    assert response.text == "OK"
+    assert "X-Response-Time" in response.headers
+    assert float(response.headers["X-Response-Time"]) >= 0
+
+
+def test_route_class():
+    response = client.post(
+        "/model/create", json={"name": "test", "description": "test"}
+    )
+    assert response.status_code == 200
+    response_json = response.json()
+    assert response_json["route_name"] == "Global.Model.create-model"
+    assert response_json["route_class"] == "TimedRoute"
+    assert response_json["router_class"] == "AppRouter"
+
+
+def test_route_name():
+    response = client.post(
+        "/model/Model001/item/create", json={"name": "test", "description": "test"}
+    )
+    assert response.status_code == 200
+    response_json = response.json()
+    assert response_json["route_name"] == "Global.Model.Item.create-item"
+    assert response_json["router_class"] == "AppRouter"