diff --git a/fastapi/dependencies/utils.py b/fastapi/dependencies/utils.py
index 84dfa4d03..d229c65c5 100644
--- a/fastapi/dependencies/utils.py
+++ b/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}"
diff --git a/fastapi/exceptions.py b/fastapi/exceptions.py
index 44d4ada86..5a7721f29 100644
--- a/fastapi/exceptions.py
+++ b/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)
diff --git a/fastapi/routing.py b/fastapi/routing.py
index 457481e32..a91205729 100644
--- a/fastapi/routing.py
+++ b/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,
diff --git a/tests/test_include_duplicate_path_route.py b/tests/test_include_duplicate_path_route.py
new file mode 100644
index 000000000..6404659d7
--- /dev/null
+++ b/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