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