PoC - named view functions

ninja/operation.py

@@ -454,44 +454,44 @@ def set_api_instance(self, api: "NinjaAPI", router: "Router") -> None:
             op.set_api_instance(api, router)
 
     def get_view(self) -> Callable:
-        view: Callable
-        if self.is_async:
-            view = self._async_view
-        else:
-            view = self._sync_view
-
-        view.__func__.csrf_exempt = True  # type: ignore
-        return view
+        is_async = self.is_async
+        operations = self.operations
+        allowed_methods = {method for op in operations for method in op.methods}
 
-    def _sync_view(self, request: HttpRequest, *a: Any, **kw: Any) -> HttpResponseBase:
-        operation = self._find_operation(request)
-        if operation is None:
-            return self._not_allowed()
-        return operation.run(request, *a, **kw)
+        if is_async:
+            from asgiref.sync import sync_to_async
 
-    async def _async_view(
-        self, request: HttpRequest, *a: Any, **kw: Any
-    ) -> HttpResponseBase:
-        from asgiref.sync import sync_to_async
-
-        operation = self._find_operation(request)
-        if operation is None:
-            return self._not_allowed()
-        if operation.is_async:
-            return await cast(AsyncOperation, operation).run(request, *a, **kw)
-        return await sync_to_async(operation.run)(request, *a, **kw)
-
-    def _find_operation(self, request: HttpRequest) -> Optional[Operation]:
-        for op in self.operations:
-            if request.method in op.methods:
-                return op
-        return None
+        def sync_view(request: HttpRequest, *a: Any, **kw: Any) -> HttpResponseBase:
+            operation = next(
+                (op for op in operations if request.method in op.methods), None
+            )
+            if operation is None:
+                return HttpResponseNotAllowed(
+                    allowed_methods, content=b"Method not allowed"
+                )
+            return operation.run(request, *a, **kw)
 
-    def _not_allowed(self) -> HttpResponse:
-        allowed_methods = set()
-        for op in self.operations:
-            allowed_methods.update(op.methods)
-        return HttpResponseNotAllowed(allowed_methods, content=b"Method not allowed")
+        async def async_view(
+            request: HttpRequest, *a: Any, **kw: Any
+        ) -> HttpResponseBase:
+            operation = next(
+                (op for op in operations if request.method in op.methods), None
+            )
+            if operation is None:
+                return HttpResponseNotAllowed(
+                    allowed_methods, content=b"Method not allowed"
+                )
+            if operation.is_async:
+                return await cast(AsyncOperation, operation).run(request, *a, **kw)
+            return await sync_to_async(operation.run)(request, *a, **kw)
+
+        view = async_view if is_async else sync_view
+        view.csrf_exempt = True  # type: ignore
+        if self.url_name:
+            view.__name__ = "".join(c if c.isalnum() else "_" for c in self.url_name)
+            if view.__name__[0].isnumeric():
+                view.__name__ = f"_{view.__name__}"
+        return view
 
 
 class ResponseObject:

tests/test_api_instance.py

@@ -9,12 +9,12 @@
 router = Router()
 
 
-@api.get("/global")
+@api.get("/global", url_name="global-op")
 def global_op(request):
     pass
 
 
-@router.get("/router")
+@router.get("/router", url_name="45")
 def router_op(request):
     pass
 
@@ -28,6 +28,16 @@ def test_api_instance():
         for path_ops in rtr.path_operations.values():
             for op in path_ops.operations:
                 assert op.api is api
+    global_op_pattern, router_op_pattern = (
+        next(
+            url_pattern
+            for url_pattern in api.urls[0]
+            if url_pattern.name == pattern_name
+        )
+        for pattern_name in ["global-op", "45"]
+    )
+    assert global_op_pattern.callback.__name__ == "global_op"
+    assert router_op_pattern.callback.__name__ == "_45"
 
 
 def test_reuse_router_error():