[Daxia] [ FastAPI ] Fix duplicate operation IDs across routers

fastapi/fastapi/openapi/utils.py

@@ -33,6 +33,7 @@
 from fastapi.utils import (
     deep_dict_update,
     generate_operation_id_for_path,
+    generate_unique_id_for_method,
     is_body_allowed_for_status_code,
 )
 from pydantic import BaseModel
@@ -233,6 +234,23 @@ def generate_operation_summary(*, route: routing.APIRoute, method: str) -> str:
     return route.name.replace("_", " ").title()
 
 
+def _get_operation_id_for_route(*, route: routing.APIRoute, method: str) -> str:
+    if route.operation_id:
+        return route.operation_id
+    if isinstance(route.generate_unique_id_function, DefaultPlaceholder):
+        return generate_unique_id_for_method(route, method)
+    return route.unique_id
+
+
+def _append_operation_id_suffix(*, operation_id: str, operation_ids: set[str]) -> str:
+    suffix = 2
+    suffixed_operation_id = f"{operation_id}_{suffix}"
+    while suffixed_operation_id in operation_ids:
+        suffix += 1
+        suffixed_operation_id = f"{operation_id}_{suffix}"
+    return suffixed_operation_id
+
+
 def get_openapi_operation_metadata(
     *, route: routing.APIRoute, method: str, operation_ids: set[str]
 ) -> dict[str, Any]:
@@ -242,14 +260,18 @@ def get_openapi_operation_metadata(
     operation["summary"] = generate_operation_summary(route=route, method=method)
     if route.description:
         operation["description"] = route.description
-    operation_id = route.operation_id or route.unique_id
+    operation_id = _get_operation_id_for_route(route=route, method=method)
     if operation_id in operation_ids:
         endpoint_name = getattr(route.endpoint, "__name__", "<unnamed_endpoint>")
         message = f"Duplicate Operation ID {operation_id} for function {endpoint_name}"
         file_name = getattr(route.endpoint, "__globals__", {}).get("__file__")
         if file_name:
             message += f" at {file_name}"
         warnings.warn(message, stacklevel=1)
+        if not route.operation_id:
+            operation_id = _append_operation_id_suffix(
+                operation_id=operation_id, operation_ids=operation_ids
+            )
     operation_ids.add(operation_id)
     operation["operationId"] = operation_id
     if route.deprecated:

fastapi/fastapi/utils.py

@@ -92,12 +92,22 @@ def generate_operation_id_for_path(
     return operation_id
 
 
+def _sanitize_operation_id_part(value: str) -> str:
+    operation_id_part = re.sub(r"[^0-9A-Za-z_]+", "_", value)
+    operation_id_part = re.sub(r"_+", "_", operation_id_part)
+    return operation_id_part.strip("_").lower()
+
+
+def generate_unique_id_for_method(route: "APIRoute", method: str) -> str:
+    path = _sanitize_operation_id_part(route.path_format)
+    name = _sanitize_operation_id_part(route.name)
+    return "_".join(part for part in (method.lower(), path, name) if part)
+
+
 def generate_unique_id(route: "APIRoute") -> str:
-    operation_id = f"{route.name}{route.path_format}"
-    operation_id = re.sub(r"\W", "_", operation_id)
     assert route.methods
-    operation_id = f"{operation_id}_{list(route.methods)[0].lower()}"
-    return operation_id
+    method = sorted(route.methods)[0]
+    return generate_unique_id_for_method(route, method)
 
 
 def deep_dict_update(main_dict: dict[Any, Any], update_dict: dict[Any, Any]) -> None:

fastapi/tests/test_operation_id_generation.py

@@ -0,0 +1,79 @@
+import pytest
+from fastapi import APIRouter, FastAPI
+from fastapi.testclient import TestClient
+
+
+def test_default_operation_ids_include_method_path_and_function_name() -> None:
+    app = FastAPI()
+    v1 = APIRouter()
+    v2 = APIRouter()
+
+    def list_users() -> dict[str, bool]:
+        return {"ok": True}
+
+    v1.add_api_route("/users", list_users, methods=["GET"])
+    v2.add_api_route("/users", list_users, methods=["GET"])
+    app.include_router(v1, prefix="/api/v1")
+    app.include_router(v2, prefix="/api/v2")
+
+    openapi = TestClient(app).get("/openapi.json").json()
+
+    assert openapi["paths"]["/api/v1/users"]["get"]["operationId"] == (
+        "get_api_v1_users_list_users"
+    )
+    assert openapi["paths"]["/api/v2/users"]["get"]["operationId"] == (
+        "get_api_v2_users_list_users"
+    )
+
+
+def test_generated_operation_ids_are_lowercase_alphanumeric_underscore() -> None:
+    app = FastAPI()
+
+    def Read_Items() -> dict[str, bool]:  # noqa: N802
+        return {"ok": True}
+
+    app.add_api_route("/API/{item_id}/details+", Read_Items, methods=["POST"])
+
+    openapi = TestClient(app).get("/openapi.json").json()
+
+    assert openapi["paths"]["/API/{item_id}/details+"]["post"]["operationId"] == (
+        "post_api_item_id_details_read_items"
+    )
+
+
+def test_multi_method_routes_get_method_specific_operation_ids() -> None:
+    app = FastAPI()
+
+    def handle_items() -> dict[str, bool]:
+        return {"ok": True}
+
+    app.add_api_route("/items", handle_items, methods=["GET", "POST"])
+
+    openapi = TestClient(app).get("/openapi.json").json()
+
+    assert openapi["paths"]["/items"]["get"]["operationId"] == (
+        "get_items_handle_items"
+    )
+    assert openapi["paths"]["/items"]["post"]["operationId"] == (
+        "post_items_handle_items"
+    )
+
+
+def test_generated_operation_id_collisions_get_numeric_suffix() -> None:
+    app = FastAPI()
+
+    def read_item() -> dict[str, bool]:
+        return {"ok": True}
+
+    app.add_api_route("/items/a-b", read_item, methods=["GET"])
+    app.add_api_route("/items/a_b", read_item, methods=["GET"])
+
+    with pytest.warns(UserWarning, match="Duplicate Operation ID"):
+        openapi = TestClient(app).get("/openapi.json").json()
+
+    assert openapi["paths"]["/items/a-b"]["get"]["operationId"] == (
+        "get_items_a_b_read_item"
+    )
+    assert openapi["paths"]["/items/a_b"]["get"]["operationId"] == (
+        "get_items_a_b_read_item_2"
+    )