Switch to decorator-based route markers

Author: goraje

fastapi_controllers/controllers.py

@@ -1,14 +1,26 @@
 import inspect
-import weakref
 from enum import Enum
 from typing import Any, Dict, List, Optional, Sequence, Union
 
 from fastapi import APIRouter, params
 
-from fastapi_controllers.definitions import HTTPRouteDefinition, RouteData, WebsocketRouteDefinition
+from fastapi_controllers.definitions import HTTPRouteMeta, Route, WebsocketRouteMeta
 from fastapi_controllers.helpers import _replace_signature, _validate_against_signature
 
 
+def _is_route(obj: Any) -> bool:
+    """
+    Check if an object is an instance of Route.
+
+    Args:
+        obj: The object to be checked.
+
+    Returns:
+        True if the checked object is an instanc of Route.
+    """
+    return isinstance(obj, Route)
+
+
 class Controller:
     prefix: str = ""
     dependencies: Optional[Sequence[params.Depends]] = None
@@ -21,7 +33,7 @@ def __init_subclass__(cls) -> None:
         for param in ["prefix", "dependencies", "tags"]:
             if not cls.__router_params__.get(param):
                 cls.__router_params__[param] = getattr(cls, param)
-        _validate_against_signature(weakref.proxy(APIRouter.__init__), kwargs=cls.__router_params__)
+        _validate_against_signature(APIRouter.__init__, kwargs=cls.__router_params__)
 
     @classmethod
     def create_router(cls) -> APIRouter:
@@ -31,24 +43,22 @@ def create_router(cls) -> APIRouter:
         Returns:
             APIRouter: An APIRouter instance.
         """
-        router = APIRouter(**cls.__router_params__)  # type: ignore
-        for _, func in inspect.getmembers(cls, predicate=inspect.isfunction):
-            _replace_signature(cls, func)
-            route_data: Optional[RouteData] = getattr(func, "__route_data__", None)
-            if route_data:
-                if isinstance(route_data.route_definition, HTTPRouteDefinition):
-                    router.add_api_route(
-                        route_data.route_args[0],
-                        func,
-                        *route_data.route_args[1:],
-                        methods=[route_data.route_definition.request_method],
-                        **route_data.route_kwargs,
-                    )
-                if isinstance(route_data.route_definition, WebsocketRouteDefinition):
-                    router.add_api_websocket_route(
-                        route_data.route_args[0],
-                        func,
-                        *route_data.route_args[1:],
-                        **route_data.route_kwargs,
-                    )
+        router = APIRouter(**(cls.__router_params__ or {}))
+        for _, route in inspect.getmembers(cls, predicate=_is_route):
+            _replace_signature(cls, route.endpoint)
+            if isinstance(route.route_meta, HTTPRouteMeta):
+                router.add_api_route(
+                    route.route_args[0],
+                    route.endpoint,
+                    *route.route_args[1:],
+                    methods=[route.route_meta.request_method],
+                    **route.route_kwargs,
+                )
+            if isinstance(route.route_meta, WebsocketRouteMeta):
+                router.add_api_websocket_route(
+                    route.route_args[0],
+                    route.endpoint,
+                    *route.route_args[1:],
+                    **route.route_kwargs,
+                )
         return router

fastapi_controllers/definitions.py

@@ -1,7 +1,6 @@
-import weakref
 from dataclasses import dataclass
 from enum import Enum
-from typing import Any, Dict, Tuple
+from typing import Any, Callable, ClassVar, Dict, Tuple
 
 from fastapi import APIRouter
 
@@ -17,36 +16,37 @@ class HTTPRequestMethod(str, Enum):
     TRACE = "TRACE"
 
 
-class RouteDefinition:
-    def __init__(self, *, binds: weakref.CallableProxyType) -> None:
+class RouteMeta:
+    def __init__(self, *, binds: Callable[..., Any]) -> None:
         self.binds = binds
 
 
-class HTTPRouteDefinition(RouteDefinition):
-    def __init__(self, *, binds: weakref.CallableProxyType, request_method: HTTPRequestMethod) -> None:
+class HTTPRouteMeta(RouteMeta):
+    def __init__(self, *, binds: Callable[..., Any], request_method: HTTPRequestMethod) -> None:
         super().__init__(binds=binds)
         self.request_method = request_method
 
 
-class WebsocketRouteDefinition(RouteDefinition):
+class WebsocketRouteMeta(RouteMeta):
     ...
 
 
 @dataclass
-class RouteData:
-    route_definition: RouteDefinition
-    route_args: Tuple[Any, ...]
-    route_kwargs: Dict[str, Any]
+class RouteMetadata:
+    delete: ClassVar[RouteMeta] = HTTPRouteMeta(binds=APIRouter.delete, request_method=HTTPRequestMethod.DELETE)
+    get: ClassVar[RouteMeta] = HTTPRouteMeta(binds=APIRouter.get, request_method=HTTPRequestMethod.GET)
+    head: ClassVar[RouteMeta] = HTTPRouteMeta(binds=APIRouter.head, request_method=HTTPRequestMethod.HEAD)
+    options: ClassVar[RouteMeta] = HTTPRouteMeta(binds=APIRouter.options, request_method=HTTPRequestMethod.OPTIONS)
+    patch: ClassVar[RouteMeta] = HTTPRouteMeta(binds=APIRouter.patch, request_method=HTTPRequestMethod.PATCH)
+    post: ClassVar[RouteMeta] = HTTPRouteMeta(binds=APIRouter.post, request_method=HTTPRequestMethod.POST)
+    put: ClassVar[RouteMeta] = HTTPRouteMeta(binds=APIRouter.put, request_method=HTTPRequestMethod.PUT)
+    trace: ClassVar[RouteMeta] = HTTPRouteMeta(binds=APIRouter.trace, request_method=HTTPRequestMethod.TRACE)
+    websocket: ClassVar[RouteMeta] = WebsocketRouteMeta(binds=APIRouter.websocket)
 
 
 @dataclass
 class Route:
-    delete = HTTPRouteDefinition(binds=weakref.proxy(APIRouter.delete), request_method=HTTPRequestMethod.DELETE)
-    get = HTTPRouteDefinition(binds=weakref.proxy(APIRouter.get), request_method=HTTPRequestMethod.GET)
-    head = HTTPRouteDefinition(binds=weakref.proxy(APIRouter.head), request_method=HTTPRequestMethod.HEAD)
-    options = HTTPRouteDefinition(binds=weakref.proxy(APIRouter.options), request_method=HTTPRequestMethod.OPTIONS)
-    patch = HTTPRouteDefinition(binds=weakref.proxy(APIRouter.patch), request_method=HTTPRequestMethod.PATCH)
-    post = HTTPRouteDefinition(binds=weakref.proxy(APIRouter.post), request_method=HTTPRequestMethod.POST)
-    put = HTTPRouteDefinition(binds=weakref.proxy(APIRouter.put), request_method=HTTPRequestMethod.PUT)
-    trace = HTTPRouteDefinition(binds=weakref.proxy(APIRouter.trace), request_method=HTTPRequestMethod.TRACE)
-    websocket = WebsocketRouteDefinition(binds=weakref.proxy(APIRouter.websocket))
+    endpoint: Callable[..., Any]
+    route_meta: RouteMeta
+    route_args: Tuple[Any, ...]
+    route_kwargs: Dict[str, Any]

fastapi_controllers/helpers.py

@@ -1,12 +1,11 @@
 import inspect
-import weakref
 from typing import Any, Callable, Dict, Optional, Tuple, Type
 
 from fastapi import Depends
 
 
 def _validate_against_signature(
-    method: weakref.CallableProxyType,
+    method: Callable[..., Any],
     args: Optional[Tuple[Any, ...]] = None,
     kwargs: Optional[Dict[str, Any]] = None,
 ) -> None:
@@ -23,14 +22,21 @@ def _validate_against_signature(
     valid_sig.bind(*(args or tuple()), **(kwargs or {}))
 
 
-def _replace_signature(cls: Type, func: Callable[..., Any]) -> None:
+def _replace_signature(klass: Type, func: Callable[..., Any]) -> None:
+    """
+    Replace the 'self' attribute with a FastAPI Depends injection.
+
+    Args:
+        klass: The class that will be injected.
+        func: The function whose signature will be replaced.
+    """
     orig_sig = inspect.signature(func)
     new_params = [
         param.replace(
             kind=inspect.Parameter.KEYWORD_ONLY,
         )
         if param.name != "self"
-        else param.replace(default=Depends(cls))
+        else param.replace(default=Depends(klass))
         for param in list(orig_sig.parameters.values())
     ]
     func.__signature__ = orig_sig.replace(parameters=new_params)  # type: ignore

fastapi_controllers/routing.py

@@ -1,28 +1,28 @@
 from typing import Any, Callable, Dict, Tuple
 
-from fastapi_controllers.definitions import Route, RouteData, RouteDefinition
+from fastapi_controllers.definitions import Route, RouteMeta, RouteMetadata
 from fastapi_controllers.helpers import _validate_against_signature
 
 
 class _RouteDecorator:
-    _route_definition: RouteDefinition
+    route_meta: RouteMeta
 
-    def __init_subclass__(cls, route_definition: RouteDefinition) -> None:
+    def __init_subclass__(cls, route_meta: RouteMeta) -> None:
         super().__init_subclass__()
-        cls._route_definition = route_definition
+        cls.route_meta = route_meta
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.route_args = args
         self.route_kwargs = kwargs
-        _validate_against_signature(self._route_definition.binds, args=args, kwargs=kwargs)
+        _validate_against_signature(self.route_meta.binds, args=args, kwargs=kwargs)
 
-    def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]:
-        func.__route_data__ = RouteData(  # type: ignore
-            route_definition=self._route_definition,
+    def __call__(self, endpoint: Callable[..., Any]) -> Route:
+        return Route(
+            endpoint=endpoint,
+            route_meta=self.route_meta,
             route_args=self.route_args,
             route_kwargs=self.route_kwargs,
         )
-        return func
 
     @property
     def route_args(self) -> Tuple[Any, ...]:
@@ -41,37 +41,37 @@ def route_kwargs(self, value: Dict[str, Any]) -> None:
         self._route_kwargs = value
 
 
-class delete(_RouteDecorator, route_definition=Route.delete):
+class delete(_RouteDecorator, route_meta=RouteMetadata.delete):
     ...
 
 
-class get(_RouteDecorator, route_definition=Route.get):
+class get(_RouteDecorator, route_meta=RouteMetadata.get):
     ...
 
 
-class head(_RouteDecorator, route_definition=Route.head):
+class head(_RouteDecorator, route_meta=RouteMetadata.head):
     ...
 
 
-class options(_RouteDecorator, route_definition=Route.options):
+class options(_RouteDecorator, route_meta=RouteMetadata.options):
     ...
 
 
-class patch(_RouteDecorator, route_definition=Route.patch):
+class patch(_RouteDecorator, route_meta=RouteMetadata.patch):
     ...
 
 
-class post(_RouteDecorator, route_definition=Route.post):
+class post(_RouteDecorator, route_meta=RouteMetadata.post):
     ...
 
 
-class put(_RouteDecorator, route_definition=Route.put):
+class put(_RouteDecorator, route_meta=RouteMetadata.put):
     ...
 
 
-class trace(_RouteDecorator, route_definition=Route.trace):
+class trace(_RouteDecorator, route_meta=RouteMetadata.trace):
     ...
 
 
-class websocket(_RouteDecorator, route_definition=Route.websocket):
+class websocket(_RouteDecorator, route_meta=RouteMetadata.websocket):
     ...

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "fastapi-controllers"
-version = "0.2.1"
+version = "0.3.0"
 description = "Simple Controller implementation for FastAPI"
 authors = ["Jerzy Góra <[email protected]>"]
 license = "MIT"

tests/unit/test_controllers.py

@@ -1,10 +1,12 @@
+from typing import Any
 from unittest.mock import MagicMock
 
 import pytest
 from fastapi import APIRouter
 from pytest_mock import MockerFixture
 
-from fastapi_controllers.controllers import Controller
+from fastapi_controllers.controllers import Controller, _is_route
+from fastapi_controllers.definitions import Route
 from fastapi_controllers.routing import get, websocket
 
 
@@ -13,6 +15,18 @@ def validator(mocker: MockerFixture) -> MagicMock:
     return mocker.patch("fastapi_controllers.controllers._validate_against_signature")
 
 
+IS_ROUTE_SCENARIOS = [
+    (Route(endpoint="TEST", route_meta="TEST", route_args="TEST", route_kwargs="TEST"), True),  # type: ignore
+    (type("NotRoute", (), {}), False),
+]
+
+
+def describe_is_route() -> None:
+    @pytest.mark.parametrize("obj,expected", IS_ROUTE_SCENARIOS)
+    def it_validates_if_obj_is_a_route(obj: Any, expected: bool) -> None:
+        assert _is_route(obj) is expected
+
+
 def describe_Controller() -> None:
     def it_sets_router_params() -> None:
         class FakeController(Controller):
@@ -34,14 +48,13 @@ class FakeController(Controller):
 
     def it_validates_apirouter_parameters(mocker: MockerFixture, validator: MagicMock) -> None:
 
-        mocked_proxy = mocker.patch("weakref.proxy", return_value="FAKE_PROXY")
+        mocked_apirouter_init = mocker.patch("fastapi_controllers.controllers.APIRouter.__init__")
 
         class _(Controller):
             ...
 
-        mocked_proxy.assert_called_once_with(APIRouter.__init__)
         validator.assert_called_once_with(
-            "FAKE_PROXY",
+            mocked_apirouter_init,
             kwargs={
                 "prefix": "",
                 "dependencies": None,
@@ -72,7 +85,7 @@ def fake_method(self) -> None:
                     ...
 
             FakeController.create_router()
-            replace.assert_called_once_with(FakeController, FakeController.fake_method)
+            replace.assert_called_once_with(FakeController, FakeController.fake_method.endpoint)
 
         def it_configures_the_router_and_routes(mocker: MockerFixture) -> None:
             apirouter = mocker.patch("fastapi_controllers.controllers.APIRouter")
@@ -92,11 +105,11 @@ def fake_ws(self) -> None:
             apirouter.assert_called_once_with(prefix="/test", dependencies=None, tags=None)
             apirouter.return_value.add_api_route.assert_called_once_with(
                 "/get",
-                FakeController.fake_method,
+                FakeController.fake_method.endpoint,
                 deprecated=True,
                 methods=["GET"],
             )
             apirouter.return_value.add_api_websocket_route.assert_called_once_with(
                 "/ws",
-                FakeController.fake_ws,
+                FakeController.fake_ws.endpoint,
             )