Merge 8ced5d0e54 into 1d434dec47

fastapi/dependencies/utils.py

@@ -12,6 +12,7 @@ from typing import (
     Mapping,
     Optional,
     Sequence,
+    Set,
     Tuple,
     Type,
     Union,
@@ -978,3 +979,8 @@ def get_body_field(
         field_info=BodyFieldInfo(**BodyFieldInfo_kwargs),
     )
     return final_field
+
+
+def get_path_hash_val(path: str, methods: Optional[Set[str]] = None) -> str:
+    methods = methods or {"GET"}
+    return f"path:{path};methods:{methods}"

fastapi/exceptions.py

@@ -174,3 +174,10 @@ class ResponseValidationError(ValidationException):
         for err in self._errors:
             message += f"  {err}\n"
         return message
+
+
+class RouteAlreadyExistsError(FastAPIError):
+    def __init__(self, f_name: str):
+        self.f_name = f_name
+        self.message = f"Route defined for {f_name} already exists!"
+        super().__init__(self.message)

fastapi/routing.py

@@ -38,6 +38,7 @@ from fastapi.dependencies.utils import (
     get_dependant,
     get_flat_dependant,
     get_parameterless_sub_dependant,
+    get_path_hash_val,
     get_typed_return_annotation,
     solve_dependencies,
 )
@@ -46,6 +47,7 @@ from fastapi.exceptions import (
     FastAPIError,
     RequestValidationError,
     ResponseValidationError,
+    RouteAlreadyExistsError,
     WebSocketRequestValidationError,
 )
 from fastapi.types import DecoratedCallable, IncEx
@@ -566,6 +568,7 @@ class APIRoute(routing.Route):
             name=self.unique_id,
             embed_body_fields=self._embed_body_fields,
         )
+        self.hash_val = get_path_hash_val(self.path, self.methods)
         self.app = request_response(self.get_route_handler())
 
     def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
@@ -858,6 +861,7 @@ class APIRouter(routing.Router):
         self.route_class = route_class
         self.default_response_class = default_response_class
         self.generate_unique_id_function = generate_unique_id_function
+        self.added_routes: Set[str] = set()
 
     def route(
         self,
@@ -958,6 +962,9 @@ class APIRouter(routing.Router):
             openapi_extra=openapi_extra,
             generate_unique_id_function=current_generate_unique_id,
         )
+        if route.hash_val in self.added_routes:
+            raise RouteAlreadyExistsError(route.name)
+        self.added_routes.add(route.hash_val)
         self.routes.append(route)
 
     def api_route(
@@ -1041,6 +1048,10 @@ class APIRouter(routing.Router):
             dependencies=current_dependencies,
             dependency_overrides_provider=self.dependency_overrides_provider,
         )
+        hash_val = get_path_hash_val(route.path)
+        if hash_val in self.added_routes:
+            raise RouteAlreadyExistsError(route.name)
+        self.added_routes.add(hash_val)
         self.routes.append(route)
 
     def websocket(
@@ -1331,6 +1342,10 @@ class APIRouter(routing.Router):
                 )
             elif isinstance(route, routing.Route):
                 methods = list(route.methods or [])
+                hash_val = get_path_hash_val(prefix + route.path, route.methods)
+                if hash_val in self.added_routes:
+                    raise RouteAlreadyExistsError(route.name)
+                self.added_routes.add(hash_val)
                 self.add_route(
                     prefix + route.path,
                     route.endpoint,

tests/test_include_duplicate_path_route.py

@@ -0,0 +1,150 @@
+import pytest
+from fastapi import APIRouter, FastAPI
+from fastapi.exceptions import RouteAlreadyExistsError
+
+
+def test_app_router_with_duplicate_path():
+    with pytest.raises(RouteAlreadyExistsError):
+        app = FastAPI()
+
+        @app.get("/items/")
+        def read_items():
+            return  # pragma: no cover
+
+        @app.get("/items/")
+        def read_items2():
+            return  # pragma: no cover
+
+
+def test_sub_with_duplicate_path():
+    with pytest.raises(RouteAlreadyExistsError):
+        app = FastAPI()
+        router = APIRouter()
+
+        @router.get("/items/")
+        def read_items():
+            return  # pragma: no cover
+
+        @router.get("/items/")
+        def read_items2():
+            return  # pragma: no cover
+
+        app.include_router(router)  # pragma: no cover
+
+
+def test_mix_app_sub_with_duplicate_path():
+    with pytest.raises(RouteAlreadyExistsError):
+        app = FastAPI()
+        router = APIRouter()
+
+        @app.get("/items/")
+        def read_items():
+            return  # pragma: no cover
+
+        @router.get("/items/")
+        def read_items2():
+            return  # pragma: no cover
+
+        app.include_router(router)  # pragma: no cover
+
+
+def test_sub_route_direct_duplicate_path():
+    with pytest.raises(RouteAlreadyExistsError):
+        app = FastAPI()
+        router = APIRouter()
+
+        @router.route("/items/")
+        def read_items():
+            return  # pragma: no cover
+
+        @router.route("/items/")
+        def read_items2():
+            return  # pragma: no cover
+
+        app.include_router(router)  # pragma: no cover
+
+
+def test_app_router_with_duplicate_path_different_method():
+    app = FastAPI()
+
+    @app.get("/items/")
+    def read_items():
+        return  # pragma: no cover
+
+    @app.post("/items/")
+    def read_items2():
+        return  # pragma: no cover
+
+
+def test_sub_with_duplicate_path_different_method():
+    app = FastAPI()
+    router = APIRouter()
+
+    @router.get("/items/")
+    def read_items():
+        return  # pragma: no cover
+
+    @router.post("/items/")
+    def read_items2():
+        return  # pragma: no cover
+
+    app.include_router(router)  # pragma: no cover
+
+
+def test_mix_app_sub_with_duplicate_different_method():
+    app = FastAPI()
+    router = APIRouter()
+
+    @app.get("/items/")
+    def read_items():
+        return  # pragma: no cover
+
+    @router.post("/items/")
+    def read_items2():
+        return  # pragma: no cover
+
+    app.include_router(router)  # pragma: no cover
+
+
+def test_sub_route_direct_duplicate_path_different_method():
+    app = FastAPI()
+    router = APIRouter()
+
+    @router.route("/items/")
+    def read_items():
+        return  # pragma: no cover
+
+    @router.route("/items/", methods=["POST"])
+    def read_items2():
+        return  # pragma: no cover
+
+    app.include_router(router)  # pragma: no cover
+
+
+def test_app_websocket_route_with_duplicate_path():
+    with pytest.raises(RouteAlreadyExistsError):
+        app = FastAPI()
+
+        @app.websocket("/items/")
+        def read_items():
+            return  # pragma: no cover
+
+        @app.websocket("/items/")
+        def read_items2():
+            return  # pragma: no cover
+
+
+def test_sub_with_duplicate_path_with_prefix():
+    with pytest.raises(RouteAlreadyExistsError):
+        app = FastAPI()
+        router = APIRouter()
+
+        @router.get("/items/")
+        def read_items():
+            return  # pragma: no cover
+
+        @router.get("/items/")
+        def read_items2():
+            return  # pragma: no cover
+
+        app.include_router(router, prefix="/prefix")  # pragma: no cover