add support for CBV method level decorators

sanic_jwt/decorators.py

@@ -5,6 +5,7 @@
 from functools import wraps
 from inspect import isawaitable
 from sanic import Blueprint
+from sanic.views import HTTPMethodView
 
 from . import exceptions
 from . import utils
@@ -58,7 +59,9 @@ async def _do_protection(*args, **kwargs):
         try:
             if instance.auth.config.do_protection():
                 (
-                    is_authenticated, status, reasons
+                    is_authenticated,
+                    status,
+                    reasons,
                 ) = instance.auth._check_authentication(
                     request, request_args=args, request_kwargs=use_kwargs
                 )
@@ -77,11 +80,11 @@ async def _do_protection(*args, **kwargs):
         except exceptions.SanicJWTException as e:
             is_authenticated = False
             status = e.status_code
-            reasons = instance.auth._reasons if (
-                instance.auth._reasons and instance.auth.config.debug()
-            ) else e.args[
-                0
-            ]
+            reasons = (
+                instance.auth._reasons
+                if (instance.auth._reasons and instance.auth.config.debug())
+                else e.args[0]
+            )
 
         if is_authenticated:
             if kwargs.get("return_response", True):
@@ -98,11 +101,11 @@ async def _do_protection(*args, **kwargs):
 
 
 def protected(initialized_on=None, **kw):
-
     def decorator(f):
-
         @wraps(f)
         async def decorated_function(request, *args, **kwargs):
+            if issubclass(request.__class__, HTTPMethodView):
+                request = args[0]
             kwargs.update(
                 {
                     "initialized_on": initialized_on,
@@ -125,9 +128,7 @@ def scoped(
     initialized_on=None,
     **kw
 ):
-
     def decorator(f):
-
         @wraps(f)
         async def decorated_function(request, *args, **kwargs):
             protect_kwargs = deepcopy(kwargs)
@@ -189,13 +190,11 @@ async def decorated_function(request, *args, **kwargs):
 
 
 def inject_user(initialized_on=None, **kw):
-
     def decorator(f):
-
         @wraps(f)
         async def decorated_function(request, *args, **kwargs):
-            if (
-                initialized_on and isinstance(initialized_on, Blueprint)
+            if initialized_on and isinstance(
+                initialized_on, Blueprint
             ):  # noqa
                 instance = initialized_on
             else:

tests/test_endpoints_cbv.py

@@ -58,8 +58,18 @@ async def get(self, request):
         return json({"protected": True})
 
 
+class PartiallyProtectedView(HTTPMethodView):
+    async def get(self, request):
+        return json({"protected": True})
+
+    @protected()
+    async def patch(self, request):
+        return json({"protected": True})
+
+
 sanic_app.add_route(PublicView.as_view(), "/")
 sanic_app.add_route(ProtectedView.as_view(), "/protected")
+sanic_app.add_route(PartiallyProtectedView.as_view(), "/partially")
 
 
 class TestEndpointsCBV(object):
@@ -71,7 +81,20 @@ def test_protected(self):
         _, response = sanic_app.test_client.get("/protected")
         assert response.status == 401
         assert response.json.get("exception") == "Unauthorized"
-        assert "Authorization header not present." in response.json.get("reasons")
+        assert "Authorization header not present." in response.json.get(
+            "reasons"
+        )
+
+    def test_partially_protected(self):
+        _, response = sanic_app.test_client.get("/partially")
+        assert response.status == 200
+
+        _, response = sanic_app.test_client.patch("/partially")
+        assert response.status == 401
+        assert response.json.get("exception") == "Unauthorized"
+        assert "Authorization header not present." in response.json.get(
+            "reasons"
+        )
 
     def test_auth_invalid_method(self):
         _, response = sanic_app.test_client.get("/auth")
@@ -83,7 +106,9 @@ def test_auth_proper_credentials(self):
             "/auth", json={"username": "user1", "password": "abcxyz"}
         )
 
-        access_token = response.json.get(sanic_jwt.config.access_token_name(), None)
+        access_token = response.json.get(
+            sanic_jwt.config.access_token_name(), None
+        )
         payload = jwt.decode(
             access_token,
             sanic_jwt.config.secret(),
@@ -97,6 +122,13 @@ def test_auth_proper_credentials(self):
         assert "exp" in payload
 
         _, response = sanic_app.test_client.get(
-            "/protected", headers={"Authorization": "Bearer {}".format(access_token)}
+            "/protected",
+            headers={"Authorization": "Bearer {}".format(access_token)},
+        )
+        assert response.status == 200
+
+        _, response = sanic_app.test_client.patch(
+            "/partially",
+            headers={"Authorization": "Bearer {}".format(access_token)},
         )
         assert response.status == 200