From 357ea1e4f762ee0d7be0240694803ca8f10bc73a Mon Sep 17 00:00:00 2001 From: Anentropic Date: Mon, 24 Nov 2025 18:57:06 +0000 Subject: [PATCH 1/4] fix for subclassed controllers sharing single route definition --- ninja_extra/controllers/base.py | 152 ++++++++++++++++++++++++++++++-- tests/test_controller.py | 59 +++++++++++++ 2 files changed, 205 insertions(+), 6 deletions(-) diff --git a/ninja_extra/controllers/base.py b/ninja_extra/controllers/base.py index 2ec563f..936bddf 100644 --- a/ninja_extra/controllers/base.py +++ b/ninja_extra/controllers/base.py @@ -4,6 +4,7 @@ import inspect import re +import types import uuid import warnings from abc import ABC @@ -84,12 +85,110 @@ def get_route_functions(cls: Type) -> Iterable[RouteFunction]: Iterable[RouteFunction]: An iterable of route functions. """ - bases = inspect.getmro(cls) - for base_cls in reversed(bases): - if base_cls not in [ControllerBase, ABC, object]: - for method in base_cls.__dict__.values(): - if hasattr(method, ROUTE_FUNCTION): - yield getattr(method, ROUTE_FUNCTION) + seen: set[str] = set() + for base_cls in inspect.getmro(cls): + if base_cls in [ControllerBase, ABC, object]: + continue + for attr_name, method in base_cls.__dict__.items(): + if attr_name in seen: + continue + if hasattr(method, ROUTE_FUNCTION): + seen.add(attr_name) + yield getattr(method, ROUTE_FUNCTION) + + +def _copy_route_enabled_function(func: Callable[..., Any]) -> Callable[..., Any]: + """ + Create a deep copy of a function object to enable per-subclass route metadata. + + When controllers inherit route-decorated methods, Python shares the same + function object across all subclasses. This causes route metadata (stored + as function attributes) to be overwritten by the last-registered controller. + + This function creates a new function object with the same code, globals, + closure, and attributes, allowing each subclass to maintain independent + route metadata. + + Args: + func: The route-decorated function to clone + + Returns: + A new function object with copied attributes (excluding ROUTE_FUNCTION) + + Note: + The ROUTE_FUNCTION attribute is intentionally excluded and will be + recreated by _attach_route_metadata() with subclass-specific metadata. + """ + new_func = types.FunctionType( + func.__code__, + func.__globals__, + name=func.__name__, + argdefs=func.__defaults__, + closure=func.__closure__, + ) + new_func.__kwdefaults__ = getattr(func, "__kwdefaults__", None) + new_func.__annotations__ = dict(getattr(func, "__annotations__", {})) + new_func.__doc__ = func.__doc__ + new_func.__module__ = func.__module__ + new_func.__qualname__ = func.__qualname__ + + # Copy over any custom attributes while skipping the route metadata – a fresh + # copy will be attached later for the subclass-specific function object. + if func.__dict__: + for key, value in func.__dict__.items(): + if key == ROUTE_FUNCTION: + continue + new_func.__dict__[key] = value + + return new_func + + +def _attach_route_metadata( + target_func: Callable[..., Any], source_route: RouteFunction +) -> None: + """ + Recreate route metadata on a cloned function by reapplying the route decorator. + + This function takes a freshly cloned function and reattaches route metadata + by invoking Route._create_route_function() with the original parameters. + This creates a new RouteFunction instance specific to the target function, + preventing metadata sharing between subclasses. + + Args: + target_func: The cloned function that needs route metadata + source_route: The original RouteFunction containing metadata to copy + + Note: + Uses a local import to avoid circular dependency issues between + controllers.base and controllers.route modules. + """ + # Local import required to avoid circular dependency + from ninja_extra.controllers.route import Route + + route = source_route.route + params = route.route_params + + Route._create_route_function( + target_func, + path=params.path, + methods=list(params.methods), + auth=params.auth, + throttle=params.throttle, + response=params.response, + operation_id=params.operation_id, + summary=params.summary, + description=params.description, + tags=list(params.tags) if params.tags else None, + deprecated=params.deprecated, + by_alias=params.by_alias, + exclude_unset=params.exclude_unset, + exclude_defaults=params.exclude_defaults, + exclude_none=params.exclude_none, + url_name=params.url_name, + include_in_schema=params.include_in_schema, + permissions=route.permissions, + openapi_extra=params.openapi_extra, + ) def get_all_controller_route_function( @@ -480,6 +579,8 @@ def __call__(self, cls: ControllerClassType) -> ControllerClassType: else: cls._api_controller = self + self._ensure_route_isolation(cls) + assert isinstance(cls.throttling_classes, (list, tuple)), ( f"Controller[{cls.__name__}].throttling_class must be a list or tuple" ) @@ -540,6 +641,45 @@ def __call__(self, cls: ControllerClassType) -> ControllerClassType: ControllerRegistry().add_controller(cls) return cls + def _ensure_route_isolation(self, cls: Type["ControllerBase"]) -> None: + """ + Ensure each controller subclass has its own copy of inherited route methods. + + When a controller inherits route-decorated methods from a base class, + Python shares the same function object across all subclasses. This causes + route metadata mutations during registration to affect all subclasses, + resulting in URL resolution pointing to the wrong controller. + + This method walks the MRO and creates per-subclass copies of inherited + route methods that haven't been explicitly overridden, ensuring each + controller maintains independent route metadata. + + Example: + >>> class BaseController(ControllerBase): + ... @http_get("/report") + ... def report(self): ... + >>> + >>> @api_controller("/alpha") + >>> class AlphaController(BaseController): + ... pass # gets its own copy of report() + """ + for base_cls in cls.__mro__[1:]: + if base_cls in (ControllerBase, ABC, object): + continue + + for attr_name, method in base_cls.__dict__.items(): + # Skip if the subclass already overrides this method + if attr_name in cls.__dict__: + continue + + if hasattr(method, ROUTE_FUNCTION): + cloned_func = _copy_route_enabled_function(method) + cloned_func.__qualname__ = f"{cls.__qualname__}.{attr_name}" + _attach_route_metadata( + cloned_func, getattr(method, ROUTE_FUNCTION) + ) + setattr(cls, attr_name, cloned_func) + @property def path_operations(self) -> Dict[str, PathView]: return self._path_operations diff --git a/tests/test_controller.py b/tests/test_controller.py index 2904de8..2826cee 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -82,6 +82,22 @@ def example(self): pass +class ReportControllerBase(ControllerBase): + @http_get("") + def report(self): + return {"controller": type(self).__name__} + + +@api_controller("/alpha") +class AlphaReportController(ReportControllerBase): + pass + + +@api_controller("/beta") +class BetaReportController(ReportControllerBase): + pass + + class TestAPIController: def test_api_controller_as_decorator(self): controller_type = api_controller("prefix", tags="new_tag", auth=FakeAuth())( @@ -321,6 +337,49 @@ async def test_controller_base_aget_object_or_none_works(self): assert isinstance(ex, exceptions.PermissionDenied) +def test_controller_subclass_routes_remain_isolated(): + api = NinjaExtraAPI() + api.register_controllers(AlphaReportController) + api.register_controllers(BetaReportController) + client = testing.TestClient(api) + + alpha_response = client.get("/alpha") + beta_response = client.get("/beta") + + assert alpha_response.status_code == 200 + assert beta_response.status_code == 200 + assert alpha_response.json() == {"controller": "AlphaReportController"} + assert beta_response.json() == {"controller": "BetaReportController"} + + +def test_controller_multi_level_inheritance_routes_isolated(): + """Test that route isolation works with multi-level inheritance.""" + # Middle layer doesn't override the route + class MiddleReportController(ReportControllerBase): + pass + + @api_controller("/gamma") + class GammaReportController(MiddleReportController): + pass + + @api_controller("/delta") + class DeltaReportController(MiddleReportController): + pass + + api = NinjaExtraAPI() + api.register_controllers(GammaReportController) + api.register_controllers(DeltaReportController) + client = testing.TestClient(api) + + gamma_response = client.get("/gamma") + delta_response = client.get("/delta") + + assert gamma_response.status_code == 200 + assert delta_response.status_code == 200 + assert gamma_response.json() == {"controller": "GammaReportController"} + assert delta_response.json() == {"controller": "DeltaReportController"} + + def test_controller_registration_through_string(): assert DisableAutoImportController.get_api_controller().registered is False From 6bc233494957059b4e15735d2b0b06a85b41ca3c Mon Sep 17 00:00:00 2001 From: Anentropic Date: Mon, 24 Nov 2025 19:25:43 +0000 Subject: [PATCH 2/4] refactor more to avoid hacky function cloning --- ninja_extra/controllers/base.py | 162 ++---------------- ninja_extra/controllers/model/builder.py | 3 +- .../controllers/route/route_functions.py | 20 +++ ninja_extra/helper.py | 17 +- 4 files changed, 51 insertions(+), 151 deletions(-) diff --git a/ninja_extra/controllers/base.py b/ninja_extra/controllers/base.py index 936bddf..4de6073 100644 --- a/ninja_extra/controllers/base.py +++ b/ninja_extra/controllers/base.py @@ -4,7 +4,6 @@ import inspect import re -import types import uuid import warnings from abc import ABC @@ -74,121 +73,31 @@ class MissingAPIControllerDecoratorException(Exception): def get_route_functions(cls: Type) -> Iterable[RouteFunction]: """ - Get all route functions from a controller class. - This function will recursively search for route functions in the base classes of the controller class - in order that they are defined. + Return fresh RouteFunction instances for a controller class. - Args: - cls (Type): The controller class. - - Returns: - Iterable[RouteFunction]: An iterable of route functions. + Each call yields a clone of the RouteFunction template stored on the + controller method, ensuring metadata is not shared across subclasses. """ + for _, method, template in _iter_route_templates(cls): + yield template.clone(method) + + +def _iter_route_templates( + cls: Type, +) -> Iterable[Tuple[str, Callable[..., Any], RouteFunction]]: seen: set[str] = set() for base_cls in inspect.getmro(cls): - if base_cls in [ControllerBase, ABC, object]: + if base_cls in (ControllerBase, ABC, object): continue for attr_name, method in base_cls.__dict__.items(): if attr_name in seen: continue - if hasattr(method, ROUTE_FUNCTION): - seen.add(attr_name) - yield getattr(method, ROUTE_FUNCTION) - - -def _copy_route_enabled_function(func: Callable[..., Any]) -> Callable[..., Any]: - """ - Create a deep copy of a function object to enable per-subclass route metadata. - - When controllers inherit route-decorated methods, Python shares the same - function object across all subclasses. This causes route metadata (stored - as function attributes) to be overwritten by the last-registered controller. - - This function creates a new function object with the same code, globals, - closure, and attributes, allowing each subclass to maintain independent - route metadata. - - Args: - func: The route-decorated function to clone - - Returns: - A new function object with copied attributes (excluding ROUTE_FUNCTION) - - Note: - The ROUTE_FUNCTION attribute is intentionally excluded and will be - recreated by _attach_route_metadata() with subclass-specific metadata. - """ - new_func = types.FunctionType( - func.__code__, - func.__globals__, - name=func.__name__, - argdefs=func.__defaults__, - closure=func.__closure__, - ) - new_func.__kwdefaults__ = getattr(func, "__kwdefaults__", None) - new_func.__annotations__ = dict(getattr(func, "__annotations__", {})) - new_func.__doc__ = func.__doc__ - new_func.__module__ = func.__module__ - new_func.__qualname__ = func.__qualname__ - - # Copy over any custom attributes while skipping the route metadata – a fresh - # copy will be attached later for the subclass-specific function object. - if func.__dict__: - for key, value in func.__dict__.items(): - if key == ROUTE_FUNCTION: + route_template = getattr(method, ROUTE_FUNCTION, None) + if route_template is None: continue - new_func.__dict__[key] = value - - return new_func - - -def _attach_route_metadata( - target_func: Callable[..., Any], source_route: RouteFunction -) -> None: - """ - Recreate route metadata on a cloned function by reapplying the route decorator. - - This function takes a freshly cloned function and reattaches route metadata - by invoking Route._create_route_function() with the original parameters. - This creates a new RouteFunction instance specific to the target function, - preventing metadata sharing between subclasses. - - Args: - target_func: The cloned function that needs route metadata - source_route: The original RouteFunction containing metadata to copy - - Note: - Uses a local import to avoid circular dependency issues between - controllers.base and controllers.route modules. - """ - # Local import required to avoid circular dependency - from ninja_extra.controllers.route import Route - - route = source_route.route - params = route.route_params - - Route._create_route_function( - target_func, - path=params.path, - methods=list(params.methods), - auth=params.auth, - throttle=params.throttle, - response=params.response, - operation_id=params.operation_id, - summary=params.summary, - description=params.description, - tags=list(params.tags) if params.tags else None, - deprecated=params.deprecated, - by_alias=params.by_alias, - exclude_unset=params.exclude_unset, - exclude_defaults=params.exclude_defaults, - exclude_none=params.exclude_none, - url_name=params.url_name, - include_in_schema=params.include_in_schema, - permissions=route.permissions, - openapi_extra=params.openapi_extra, - ) + seen.add(attr_name) + yield attr_name, method, route_template def get_all_controller_route_function( @@ -579,8 +488,6 @@ def __call__(self, cls: ControllerClassType) -> ControllerClassType: else: cls._api_controller = self - self._ensure_route_isolation(cls) - assert isinstance(cls.throttling_classes, (list, tuple)), ( f"Controller[{cls.__name__}].throttling_class must be a list or tuple" ) @@ -641,45 +548,6 @@ def __call__(self, cls: ControllerClassType) -> ControllerClassType: ControllerRegistry().add_controller(cls) return cls - def _ensure_route_isolation(self, cls: Type["ControllerBase"]) -> None: - """ - Ensure each controller subclass has its own copy of inherited route methods. - - When a controller inherits route-decorated methods from a base class, - Python shares the same function object across all subclasses. This causes - route metadata mutations during registration to affect all subclasses, - resulting in URL resolution pointing to the wrong controller. - - This method walks the MRO and creates per-subclass copies of inherited - route methods that haven't been explicitly overridden, ensuring each - controller maintains independent route metadata. - - Example: - >>> class BaseController(ControllerBase): - ... @http_get("/report") - ... def report(self): ... - >>> - >>> @api_controller("/alpha") - >>> class AlphaController(BaseController): - ... pass # gets its own copy of report() - """ - for base_cls in cls.__mro__[1:]: - if base_cls in (ControllerBase, ABC, object): - continue - - for attr_name, method in base_cls.__dict__.items(): - # Skip if the subclass already overrides this method - if attr_name in cls.__dict__: - continue - - if hasattr(method, ROUTE_FUNCTION): - cloned_func = _copy_route_enabled_function(method) - cloned_func.__qualname__ = f"{cls.__qualname__}.{attr_name}" - _attach_route_metadata( - cloned_func, getattr(method, ROUTE_FUNCTION) - ) - setattr(cls, attr_name, cloned_func) - @property def path_operations(self) -> Dict[str, PathView]: return self._path_operations diff --git a/ninja_extra/controllers/model/builder.py b/ninja_extra/controllers/model/builder.py index 3f5dff2..30264ea 100644 --- a/ninja_extra/controllers/model/builder.py +++ b/ninja_extra/controllers/model/builder.py @@ -51,7 +51,8 @@ def __init__( ) def _add_to_controller(self, func: t.Callable) -> None: - route_function = getattr(func, ROUTE_FUNCTION) + route_template = getattr(func, ROUTE_FUNCTION) + route_function = route_template.clone(func) route_function.api_controller = self._api_controller_instance self._api_controller_instance.add_controller_route_function(route_function) diff --git a/ninja_extra/controllers/route/route_functions.py b/ninja_extra/controllers/route/route_functions.py index b115acf..da3e30e 100644 --- a/ninja_extra/controllers/route/route_functions.py +++ b/ninja_extra/controllers/route/route_functions.py @@ -104,6 +104,26 @@ def as_view( as_view.get_route_function = lambda: self # type:ignore return as_view + def clone(self, view_func: Callable[..., Any]) -> "RouteFunction": + from ninja_extra.controllers.route import Route + + route_params = self.route.route_params.dict() + permissions = self.route.permissions + if permissions is not None: + permissions = list(permissions) + + if route_params["tags"] is not None: + route_params["tags"] = list(route_params["tags"]) + route_params["methods"] = list(route_params["methods"]) + + cloned_route = Route( + view_func, + **route_params, + permissions=permissions, + ) + + return type(self)(route=cloned_route) + def _process_view_function_result(self, result: Any) -> Any: """ This process any a returned value from view_func diff --git a/ninja_extra/helper.py b/ninja_extra/helper.py index f4b4de9..b9a1f96 100644 --- a/ninja_extra/helper.py +++ b/ninja_extra/helper.py @@ -15,6 +15,17 @@ def get_function_name(func_class: t.Any) -> str: @t.no_type_check def get_route_function(func: t.Callable) -> t.Optional["RouteFunction"]: - if hasattr(func, ROUTE_FUNCTION): - return func.__dict__[ROUTE_FUNCTION] - return None # pragma: no cover + controller_instance = getattr(func, "__self__", None) + + if controller_instance is not None: + controller_class = controller_instance.__class__ + api_controller = controller_class.get_api_controller() + return api_controller._controller_class_route_functions.get(func.__name__) + + # Unbound function – return a clone of the template for introspection + underlying_func = getattr(func, "__func__", func) + route_template = getattr(underlying_func, ROUTE_FUNCTION, None) + if route_template is None: + return None # pragma: no cover + + return route_template.clone(underlying_func) From 7f3b8a9e6d78d8bc19eb78c91ee755256847d00e Mon Sep 17 00:00:00 2001 From: Anentropic Date: Mon, 24 Nov 2025 20:35:34 +0000 Subject: [PATCH 3/4] fix tests for namespaced subclassed controller --- ninja_extra/testing/client.py | 17 ++++++++++++++++- tests/test_controller.py | 4 ++-- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/ninja_extra/testing/client.py b/ninja_extra/testing/client.py index 2b313e0..50ace2c 100644 --- a/ninja_extra/testing/client.py +++ b/ninja_extra/testing/client.py @@ -1,8 +1,9 @@ from json import dumps as json_dumps -from typing import Any, Callable, Dict, Optional, Type, Union +from typing import Any, Callable, Dict, Optional, Tuple, Type, Union from unittest.mock import Mock from urllib.parse import urlencode +from django.urls import Resolver404 from ninja import NinjaAPI, Router from ninja.responses import NinjaJSONEncoder from ninja.testing.client import NinjaClientBase, NinjaResponse @@ -42,6 +43,20 @@ def request( ) return self._call(func, request, kwargs) # type: ignore + def _resolve( + self, method: str, path: str, data: Dict, request_params: Any + ) -> Tuple[Callable, Mock, Dict]: + url_path = path.split("?")[0].lstrip("/") + for url in self.urls: + try: + match = url.resolve(url_path) + except Resolver404: + continue + if match: + request = self._build_request(method, path, data, request_params) + return match.func, request, match.kwargs + raise Exception(f'Cannot resolve "{path}"') + class TestClient(NinjaExtraClientBase): def _call(self, func: Callable, request: Mock, kwargs: Dict) -> "NinjaResponse": diff --git a/tests/test_controller.py b/tests/test_controller.py index 2826cee..9295db6 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -88,12 +88,12 @@ def report(self): return {"controller": type(self).__name__} -@api_controller("/alpha") +@api_controller("/alpha", urls_namespace="alpha") class AlphaReportController(ReportControllerBase): pass -@api_controller("/beta") +@api_controller("/beta", urls_namespace="beta") class BetaReportController(ReportControllerBase): pass From a7738252d554fa9d79a73fac387550d17777c2bb Mon Sep 17 00:00:00 2001 From: Anentropic Date: Mon, 24 Nov 2025 20:57:41 +0000 Subject: [PATCH 4/4] fix typing error for mypy --- .../controllers/route/route_functions.py | 26 ++++++++++++++++--- 1 file changed, 22 insertions(+), 4 deletions(-) diff --git a/ninja_extra/controllers/route/route_functions.py b/ninja_extra/controllers/route/route_functions.py index da3e30e..e17d92b 100644 --- a/ninja_extra/controllers/route/route_functions.py +++ b/ninja_extra/controllers/route/route_functions.py @@ -2,7 +2,18 @@ import warnings from contextlib import contextmanager from functools import wraps -from typing import TYPE_CHECKING, Any, Callable, Iterator, Optional, Tuple, cast +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Iterator, + List, + Optional, + Tuple, + Type, + Union, + cast, +) from django.http import HttpRequest, HttpResponse @@ -16,6 +27,11 @@ from ninja_extra.controllers.base import APIController, ControllerBase from ninja_extra.controllers.route import Route from ninja_extra.operation import Operation + from ninja_extra.permissions import BasePermission + +RoutePermissions = Optional[ + List[Union[Type["BasePermission"], "BasePermission", Any]] +] class RouteFunctionContext: @@ -108,9 +124,11 @@ def clone(self, view_func: Callable[..., Any]) -> "RouteFunction": from ninja_extra.controllers.route import Route route_params = self.route.route_params.dict() - permissions = self.route.permissions - if permissions is not None: - permissions = list(permissions) + permissions: RoutePermissions + if self.route.permissions is None: + permissions = None + else: + permissions = cast(RoutePermissions, list(self.route.permissions)) if route_params["tags"] is not None: route_params["tags"] = list(route_params["tags"])