Merge pull request #7 from eadwinCode/dynamic_config

Dynamic Config

Author: eadwinCode

README.md

@@ -196,16 +196,16 @@ JSON Encoder class that will be used by the `PYJWT` to encode the `jwt_payload`.
 
 The `JwtService` uses [PYJWT](https://pypi.org/project/PyJWT/) underneath.
 
-### _jwt_service.sign(payload: dict, headers: Dict[str, t.Any] = None) -> str_
-Creates a jwt token for the provided payload
+### _jwt_service.sign(payload: dict, headers: Dict[str, t.Any] = None, **jwt_config: t.Any) -> str_
+Creates a jwt token for the provided payload. Also, you can override default jwt config by using passing some keyword argument as a `jwt_config`
 
-### _jwt_service.sign_async(payload: dict, headers: Dict[str, t.Any] = None) -> str_
+### _jwt_service.sign_async(payload: dict, headers: Dict[str, t.Any] = None, **jwt_config: t.Any) -> str_
 Async action for `jwt_service.sign`
 
-### _jwt_service.decode(token: str, verify: bool = True) -> t.Dict[str, t.Any]:_
+### _jwt_service.decode(token: str, verify: bool = True, **jwt_config: t.Any) -> t.Dict[str, t.Any]:_
 Verifies and decodes provided token. And raises JWTException exception if token is invalid or expired
 
-### _jwt_service.decode_async(token: str, verify: bool = True) -> t.Dict[str, t.Any]:_
+### _jwt_service.decode_async(token: str, verify: bool = True, **jwt_config: t.Any) -> t.Dict[str, t.Any]:_
 Async action for `jwt_service.decode`
 
 

ellar_jwt/services.py

@@ -18,55 +18,68 @@ class JWTService:
     def __init__(self, jwt_config: JWTConfiguration) -> None:
         self.jwt_config = jwt_config
 
-        self.jwks_client = (
-            PyJWKClient(self.jwt_config.jwk_url) if self.jwt_config.jwk_url else None
-        )
-        self.leeway = self.jwt_config.leeway
+    def get_jwks_client(self, jwt_config: JWTConfiguration) -> t.Optional[PyJWKClient]:
+        jwks_client = PyJWKClient(jwt_config.jwk_url) if jwt_config.jwk_url else None
+        return jwks_client
 
-    def get_leeway(self) -> timedelta:
-        if self.leeway is None:
+    def get_leeway(self, jwt_config: JWTConfiguration) -> timedelta:
+        if jwt_config.leeway is None:
             return timedelta(seconds=0)
-        elif isinstance(self.leeway, (int, float)):
-            return timedelta(seconds=self.leeway)
-        elif isinstance(self.leeway, timedelta):
-            return self.leeway
+        elif isinstance(jwt_config.leeway, (int, float)):
+            return timedelta(seconds=jwt_config.leeway)
+        elif isinstance(jwt_config.leeway, timedelta):
+            return jwt_config.leeway
 
-    def get_verifying_key(self, token: t.Any) -> bytes:
+    def get_verifying_key(self, token: t.Any, jwt_config: JWTConfiguration) -> bytes:
         if self.jwt_config.algorithm.startswith("HS"):
-            return self.jwt_config.signing_secret_key.encode()
+            return jwt_config.signing_secret_key.encode()
 
-        if self.jwks_client:
+        jwks_client = self.get_jwks_client(jwt_config)
+        if jwks_client:
             try:
-                p_jwk = self.jwks_client.get_signing_key_from_jwt(token)
+                p_jwk = jwks_client.get_signing_key_from_jwt(token)
                 return p_jwk.key  # type:ignore[no-any-return]
             except PyJWKClientError as ex:
                 raise JWTTokenException("Token is invalid or expired") from ex
 
-        return self.jwt_config.verifying_secret_key.encode()
+        return jwt_config.verifying_secret_key.encode()
+
+    def _merge_configurations(self, **jwt_config: t.Any) -> JWTConfiguration:
+        jwt_config_default = self.jwt_config.dict()
+        jwt_config_default.update(jwt_config)
+        return JWTConfiguration(**jwt_config_default)
 
     def sign(
-        self, payload: dict, headers: t.Optional[t.Dict[str, t.Any]] = None
+        self,
+        payload: dict,
+        headers: t.Optional[t.Dict[str, t.Any]] = None,
+        **jwt_config: t.Any,
     ) -> str:
         """
         Returns an encoded token for the given payload dictionary.
         """
-
-        jwt_payload = Token(jwt_config=self.jwt_config).build(payload.copy())
+        _jwt_config = self._merge_configurations(**jwt_config)
+        jwt_payload = Token(jwt_config=_jwt_config).build(payload.copy())
 
         return jwt.encode(
             jwt_payload,
-            self.jwt_config.signing_secret_key,
-            algorithm=self.jwt_config.algorithm,
-            json_encoder=self.jwt_config.json_encoder,
+            _jwt_config.signing_secret_key,
+            algorithm=_jwt_config.algorithm,
+            json_encoder=_jwt_config.json_encoder,
             headers=headers,
         )
 
     async def sign_async(
-        self, payload: dict, headers: t.Optional[t.Dict[str, t.Any]] = None
+        self,
+        payload: dict,
+        headers: t.Optional[t.Dict[str, t.Any]] = None,
+        **jwt_config: t.Any,
     ) -> str:
-        return await anyio.to_thread.run_sync(self.sign, payload, headers)
+        return await anyio.to_thread.run_sync(self.sign, payload, headers, **jwt_config)
 
-    def decode(self, token: str, verify: bool = True) -> t.Dict[str, t.Any]:
+    def decode(
+        self, token: str, verify: bool = True, **jwt_config: t.Any
+    ) -> t.Dict[str, t.Any]:
         """
         Performs a validation of the given token and returns its payload
         dictionary.
@@ -75,15 +88,16 @@ def decode(self, token: str, verify: bool = True) -> t.Dict[str, t.Any]:
         signature check fails, or if its 'exp' claim indicates it has expired.
         """
         try:
+            _jwt_config = self._merge_configurations(**jwt_config)
             return jwt.decode(  # type: ignore[no-any-return]
                 token,
-                self.get_verifying_key(token),
-                algorithms=[self.jwt_config.algorithm],
-                audience=self.jwt_config.audience,
-                issuer=self.jwt_config.issuer,
-                leeway=self.get_leeway(),
+                self.get_verifying_key(token, _jwt_config),
+                algorithms=[_jwt_config.algorithm],
+                audience=_jwt_config.audience,
+                issuer=_jwt_config.issuer,
+                leeway=self.get_leeway(_jwt_config),
                 options={
-                    "verify_aud": self.jwt_config.audience is not None,
+                    "verify_aud": _jwt_config.audience is not None,
                     "verify_signature": verify,
                 },
             )
@@ -92,5 +106,7 @@ def decode(self, token: str, verify: bool = True) -> t.Dict[str, t.Any]:
         except InvalidTokenError as ex:
             raise JWTTokenException("Token is invalid or expired") from ex
 
-    async def decode_async(self, token: str, verify: bool = True) -> t.Dict[str, t.Any]:
-        return await anyio.to_thread.run_sync(self.decode, token, verify)
+    async def decode_async(
+        self, token: str, verify: bool = True, **jwt_config: t.Any
+    ) -> t.Dict[str, t.Any]:
+        return await anyio.to_thread.run_sync(self.decode, token, verify, **jwt_config)