Support controller reuse across multiple NinjaExtraAPI instances

ninja_extra/main.py

@@ -1,3 +1,4 @@
+import logging
 from importlib import import_module
 from typing import (
     Any,
@@ -11,6 +12,7 @@
     Union,
     cast,
 )
+from weakref import WeakSet
 
 from django.core.exceptions import ImproperlyConfigured
 from django.http import HttpRequest, HttpResponse
@@ -33,6 +35,7 @@
     "NinjaExtraAPI",
 ]
 
+logger = logging.getLogger(__name__)
 
 class NinjaExtraAPI(NinjaAPI):
     def __init__(
@@ -79,6 +82,8 @@ def __init__(
         self._routers: List[Tuple[str, router.Router]] = []  # type: ignore
         self.default_router = router.Router()
         self.add_router("", self.default_router)
+        self._registered_controllers: "WeakSet[type[ControllerBase]]" = WeakSet()
+        self._controller_clones: dict[type[ControllerBase], type[ControllerBase]] = {}
 
     def api_exception_handler(
         self, request: HttpRequest, exc: exceptions.APIException
@@ -120,11 +125,45 @@ def register_controllers(
                 raise ImproperlyConfigured(
                     f"{controller.__class__.__name__} class is not a controller"
                 )
+            if controller in self._registered_controllers or controller in self._controller_clones:
+                continue
+
             api_controller: APIController = controller.get_api_controller()
+            if api_controller.registered:
+                # Clone the controller for isolation in this API instance
+                # Create a unique subclass to avoid shared class state
+                cloned_controller_name = f"{controller.__name__}_clone_for_{self.urls_namespace or 'api'}"
+                cloned_controller = type(cloned_controller_name, (controller,), {})
+
+                # Clone the APIController config from the original
+                cloned_api_controller = APIController(
+                    prefix=api_controller.prefix,
+                    auth=api_controller.auth if api_controller.auth is not NOT_SET else NOT_SET,
+                    throttle=api_controller.throttle if api_controller.throttle is not NOT_SET else NOT_SET,
+                    tags=api_controller.tags,
+                    permissions=api_controller.permission_classes,
+                    auto_import=api_controller.auto_import,
+                )
+
+                # Apply the cloned decorator to the cloned class (this rebuilds routes, operations, etc.)
+                cloned_controller = cloned_api_controller(cloned_controller)
+                logger.info(
+                    "Controller cloned %s from %s at namespace=%s from app=%s",
+                    cloned_controller.__name__,
+                    controller.__name__,
+                    self.urls_namespace or "api",
+                    getattr(self, "app_name", "ninja"),
+                )
+                # Update to use the cloned versions for registration
+                api_controller = cloned_api_controller
+                self._registered_controllers.add(controller)
+                self._controller_clones[controller] = cloned_controller
+                controller = cloned_controller  # Optional, but ensures consistency
             if not api_controller.registered:
                 self._routers.extend(api_controller.build_routers())  # type: ignore
                 api_controller.set_api_instance(self)
                 api_controller.registered = True
+                self._registered_controllers.add(controller)
 
     def auto_discover_controllers(self) -> None:
         from django.apps import apps

tests/test_api_instance.py

@@ -1,3 +1,4 @@
+from typing import Optional
 from unittest import mock
 
 import pytest
@@ -83,3 +84,91 @@ def example(self):
     res = client.get("/another/example")
     assert res.status_code == 200
     assert res.content == b'"Create Response Works"'
+
+
+def test_same_controller_two_apis_works():
+    @api_controller("/ping")
+    class P:
+        @http_get("")
+        def ping(self): return {"ok": True}
+
+    a = NinjaExtraAPI(urls_namespace="a")
+    b = NinjaExtraAPI(urls_namespace="b")
+
+    a.register_controllers(P)
+    b.register_controllers(P)  # triggers clone path
+
+    assert TestClient(a).get("/ping").json() == {"ok": True}
+    assert TestClient(b).get("/ping").json() == {"ok": True}
+
+    ra = dict(a._routers)["/ping"]
+    rb = dict(b._routers)["/ping"]
+    # Different Router objects per API (isolation)
+    assert ra is not rb
+
+
+def test_openapi_schema_params_are_correct_on_two_apis():
+    @api_controller("/")
+    class ItemsController:
+        @http_get("/items_1")
+        def items_1(self, ordering: Optional[str] = None):
+            return {"ok": True}
+
+    # Two independent API instances
+    api_a = NinjaExtraAPI(title="A")
+    api_b = NinjaExtraAPI(title="B")
+
+    api_a.register_controllers(ItemsController)
+    api_b.register_controllers(ItemsController)
+
+    expected_params = [
+        {
+            "in": "query",
+            "name": "ordering",
+            "required": False,
+            "schema": {
+                "anyOf": [{"type": "string"}, {"type": "null"}],
+                "title": "Ordering",
+            },
+        }
+    ]
+
+    # Check API A schema
+    schema_a = api_a.get_openapi_schema()
+    op_a = schema_a["paths"]["/api/items_1"]["get"]
+    assert op_a["parameters"] == expected_params
+
+    # Check API B schema
+    schema_b = api_b.get_openapi_schema()
+    op_b = schema_b["paths"]["/api/items_1"]["get"]
+    assert op_b["parameters"] == expected_params
+
+    # (Optional) also confirm the route actually works on both APIs
+    ca = TestClient(api_a)
+    cb = TestClient(api_b)
+    assert ca.get("/items_1").status_code == 200
+    assert cb.get("/items_1").status_code == 200
+
+
+def test_clone_is_cached_per_api_not_recreated():
+    """Register the same original class twice on the same API -> reuse cached clone, no new routers."""
+    @api_controller("/x")
+    class X:
+        @http_get("")
+        def ok(self): return {"ok": True}
+
+    a = NinjaExtraAPI(urls_namespace="a")
+    b = NinjaExtraAPI(urls_namespace="b")
+
+    # Mount on A (original)
+    a.register_controllers(X)
+    # Mount on B (clone)
+    b.register_controllers(X)
+    # Re-register same original on B (should reuse the cached clone; no new routers added)
+    before = len(b._routers)
+    b.register_controllers(X)
+    after = len(b._routers)
+    assert before == after
+
+    # Optional: ensure path exists and works
+    assert TestClient(b).get("/x").json() == {"ok": True}