refactored check user authentication rule (#27)

Author: eadwinCode

ninja_jwt/__init__.py

@@ -1,3 +1,3 @@
 """Django Ninja JWT - JSON Web Token for Django-Ninja"""
 
-__version__ = "5.2.4"
+__version__ = "5.2.5"

ninja_jwt/controller.py

@@ -66,6 +66,7 @@ class TokenObtainPairController(ControllerBase):
         url_name="token_obtain_pair",
     )
     def obtain_token(self, user_token: schema.obtain_pair_schema):
+        user_token.check_user_authentication_rule()
         return user_token.to_response_schema()
 
     @http_post(
@@ -86,6 +87,7 @@ class TokenObtainSlidingController(TokenObtainPairController):
         url_name="token_obtain_sliding",
     )
     def obtain_token(self, user_token: schema.obtain_sliding_schema):
+        user_token.check_user_authentication_rule()
         return user_token.to_response_schema()
 
     @http_post(
@@ -146,6 +148,7 @@ class AsyncTokenObtainPairController(TokenObtainPairController):
             url_name="token_obtain_pair",
         )
         async def obtain_token(self, user_token: schema.obtain_pair_schema):
+            await sync_to_async(user_token.check_user_authentication_rule)()
             return user_token.to_response_schema()
 
         @http_post(
@@ -164,6 +167,7 @@ class AsyncTokenObtainSlidingController(TokenObtainSlidingController):
             url_name="token_obtain_sliding",
         )
         async def obtain_token(self, user_token: schema.obtain_sliding_schema):
+            await sync_to_async(user_token.check_user_authentication_rule)()
             return user_token.to_response_schema()
 
         @http_post(

ninja_jwt/schema.py

@@ -1,5 +1,5 @@
 import warnings
-from typing import Any, Callable, Dict, Optional, Type, Union, cast
+from typing import Any, Callable, Dict, Optional, Type, cast
 
 from django.conf import settings
 from django.contrib.auth import authenticate, get_user_model
@@ -46,11 +46,11 @@ class TokenInputSchemaMixin(InputSchemaMixin):
         "no_active_account": _("No active account found with the given credentials")
     }
 
-    @classmethod
-    def check_user_authentication_rule(
-        cls, user: Optional[Union[AbstractUser, Any]], values: Dict
-    ) -> bool:
-        return api_settings.USER_AUTHENTICATION_RULE(user)
+    def check_user_authentication_rule(self) -> None:
+        if not api_settings.USER_AUTHENTICATION_RULE(self._user):
+            raise exceptions.AuthenticationFailed(
+                self._default_error_messages["no_active_account"]
+            )
 
     @classmethod
     def validate_values(cls, values: Dict) -> dict:
@@ -71,14 +71,13 @@ def validate_values(cls, values: Dict) -> dict:
             raise exceptions.ValidationError({"password": "password is required"})
 
         _user = authenticate(**values)
+        cls._user = _user
 
-        if not cls.check_user_authentication_rule(_user, values):
+        if _user is None:
             raise exceptions.AuthenticationFailed(
                 cls._default_error_messages["no_active_account"]
             )
 
-        cls._user = _user
-
         return values
 
     def output_schema(self) -> Schema:
@@ -117,6 +116,7 @@ def post_validate_schema(cls, values: Dict) -> dict:
         :return:
         """
         # get_token can return values that wants to apply to `OutputSchema`
+
         data = cls.get_token(cls._user)
 
         if not isinstance(data, dict):

tests/test_custom_schema.py

@@ -61,6 +61,10 @@ def get_token(cls, user) -> Dict:
 class MyNewObtainTokenSlidingSchemaInput(TokenObtainSlidingInputSchema):
     my_extra_field: str
 
+    def check_user_authentication_rule(self) -> bool:
+        my_extra_field = self.my_extra_field
+        return api_settings.USER_AUTHENTICATION_RULE(self._user)
+
     @classmethod
     def get_response_schema(cls):
         return MyNewObtainTokenSlidingSchemaOutput

tests/utils.py

@@ -1,9 +1,8 @@
-import contextlib
 import json
 
 from django.db import connection
 from django.db.migrations.executor import MigrationExecutor
-from django.test import Client, TestCase
+from django.test import Client
 from django.test.utils import override_settings
 
 from ninja_jwt.compat import reverse