Small refactoring, wip async, docs improve

Author: k4black

docs/docs/developing.md

@@ -2,7 +2,7 @@
 
 Install dev dependencies 
 ```shell
-python -m pip install .[docs]  # \[docs\] in zsh
+python -m pip install .[docs,test]  # \[docs,test\] in zsh
 ```
 
 
@@ -69,11 +69,11 @@ Edit it in `docs/`
 mkdocs serve --config-file docs/mkdocs.yml
 ```
 
-Note: Server will auto-restart for all changed `docs/*` files. 
+Note: Server will auto-restart for all changed `docs/*` files.  
 If you want to edit `README.md` or `CONTRIBUTING.md` you should restart server on each change.  
 
 
-### Building (`TODO`)
+### Building pkg docs (`TODO`)
 
 Add python backend docs `TODO`
 ```shell

docs/docs/user_guide/classes.md

@@ -0,0 +1,52 @@
+# Classes
+
+This library made in fastapi style, so it can be used as standard security features 
+
+
+## Security classes 
+
+#### Credentials 
+
+* `JwtAuthorizationCredentials` - universal credentials for access and refresh tokens.  
+    Provide access to subject and unique token identifier (jti)
+    ```python
+    def foo(credentials: JwtAuthorizationCredentials = Security(access_security)):
+        return credentials["username"], credentials.jti
+    ```
+
+#### Access tokens
+
+* `JwtAccessBearer` - read access token from bearer header only
+* `JwtAccessCookie` - read access token from cookies only
+* `JwtAccessBearerCookie` - read access token from both bearer and cookie
+
+#### Refresh tokens
+
+* `JwtRefreshBearer` - read access token from bearer header only
+* `JwtRefreshCookie` - read access token from cookies only
+* `JwtRefreshBearerCookie` - read access token from both bearer and cookie
+
+
+### Create
+
+You can create `access_security` / `refresh_security` in multiple ways 
+```python
+# Manually
+access_security = JwtAccessBearerCookie(
+    secret_key="other_secret_key",
+    auto_error=True,
+    access_expires_delta=timedelta(hours=1),  # custom access token valid timedelta
+    refresh_expires_delta=timedelta(days=1),  # custom access token valid timedelta
+)
+
+# Create from another object, copy all params
+refresh_security = JwtRefreshBearer.from_other(access_security)
+
+# Create from another object, rewrite some params
+other_access_security = JwtAccessCookie.from_other(
+    access_security, 
+    secret_key='!key!', 
+    auto_error=False
+)
+```
+

docs/docs/user_guide.md docs/docs/user_guide/examples.mddocs/docs/user_guide.md renamed to docs/docs/user_guide/examples.md

@@ -1,9 +1,9 @@
-# User guide 
+# Examples
 
 This library made in fastapi style, so it can be used as standard security features 
 
 
-## Example 
+## Full Example 
 
 
 ```python 

docs/mkdocs.yml

@@ -16,7 +16,9 @@ edit_uri: ""
 
 nav:
     - Home: index.md
-    - User Guide: user_guide.md
+    - User Guide:
+        - Classes: user_guide/classes.md
+        - Examples: user_guide/examples.md
     - Developing: developing.md
     - Release notes: release-notes.md
 

fastapi_jwt/jwt.py

@@ -1,7 +1,7 @@
 from abc import ABC
 from datetime import datetime, timedelta
 from typing import Any, Dict, Optional, Set
-from uuid import uuid1
+from uuid import uuid4
 
 from fastapi.exceptions import HTTPException
 from fastapi.param_functions import Security
@@ -83,6 +83,24 @@ def __init__(
         self.access_expires_delta = access_expires_delta or timedelta(minutes=15)
         self.refresh_expires_delta = refresh_expires_delta or timedelta(days=31)
 
+    @classmethod
+    def from_other(
+        cls,
+        other: 'JwtAuthBase',
+        secret_key: Optional[str] = None,
+        auto_error: Optional[bool] = None,
+        algorithm: Optional[str] = None,
+        access_expires_delta: Optional[timedelta] = None,
+        refresh_expires_delta: Optional[timedelta] = None,
+    ) -> 'JwtAuthBase':
+        return cls(
+            secret_key=secret_key or other.secret_key,
+            auto_error=auto_error or other.auto_error,
+            algorithm=algorithm or other.algorithm,
+            access_expires_delta=access_expires_delta or other.access_expires_delta,
+            refresh_expires_delta=refresh_expires_delta or other.refresh_expires_delta,
+        )
+
     def _decode(self, token: str) -> Optional[Dict[str, Any]]:
         try:
             payload: Dict[str, Any] = jwt.decode(
@@ -124,24 +142,17 @@ def _generate_payload(
             "jti": unique_identifier,  # uuid
         }
 
-    def _get_token(
-        self,
-        bearer: Optional[HTTPBearer] = None,
-        cookie: Optional[APIKeyCookie] = None,
-    ) -> Optional[str]:
-        if bearer:
-            return str(bearer.credentials)  # type: ignore
-        if cookie:
-            return str(cookie)
-        return None
-
-    def _get_payload(
+    async def _get_payload(
         self, bearer: Optional[HTTPBearer], cookie: Optional[APIKeyCookie]
     ) -> Optional[Dict[str, Any]]:
-        refresh_token = self._get_token(bearer, cookie)  # TODO: del function
+        token: Optional[str] = None
+        if bearer:
+            token = str(bearer.credentials)  # type: ignore
+        elif cookie:
+            token = str(cookie)
 
         # Check token exist
-        if refresh_token is None:
+        if not token:
             if self.auto_error:
                 raise HTTPException(
                     status_code=HTTP_401_UNAUTHORIZED, detail="Credentials are not provided"
@@ -150,7 +161,7 @@ def _get_payload(
                 return None
 
         # Try to decode jwt token. auto_error on error
-        payload = self._decode(refresh_token)
+        payload = self._decode(token)
         return payload
 
     def create_access_token(
@@ -160,7 +171,7 @@ def create_access_token(
         unique_identifier: Optional[str] = None,
     ) -> str:
         expires_delta = expires_delta or self.access_expires_delta
-        unique_identifier = unique_identifier or str(uuid1())
+        unique_identifier = unique_identifier or str(uuid4())
         to_encode = self._generate_payload(
             subject, expires_delta, unique_identifier, "access"
         )
@@ -177,7 +188,7 @@ def create_refresh_token(
         unique_identifier: Optional[str] = None,
     ) -> str:
         expires_delta = expires_delta or self.refresh_expires_delta
-        unique_identifier = unique_identifier or str(uuid1())
+        unique_identifier = unique_identifier or str(uuid4())
         to_encode = self._generate_payload(
             subject, expires_delta, unique_identifier, "refresh"
         )
@@ -252,12 +263,12 @@ def __init__(
             refresh_expires_delta=refresh_expires_delta,
         )
 
-    def _get_credentials(
+    async def _get_credentials(
         self,
         bearer: Optional[JwtAuthBase.JwtAccessBearer],
         cookie: Optional[JwtAuthBase.JwtAccessCookie],
     ) -> Optional[JwtAuthorizationCredentials]:
-        payload = self._get_payload(bearer, cookie)
+        payload = await self._get_payload(bearer, cookie)
 
         if payload:
             return JwtAuthorizationCredentials(
@@ -284,10 +295,10 @@ def __init__(
             refresh_expires_delta=refresh_expires_delta,
         )
 
-    def __call__(
+    async def __call__(
         self, bearer: JwtAuthBase.JwtAccessBearer = Security(JwtAccess._bearer)
     ) -> Optional[JwtAuthorizationCredentials]:
-        return self._get_credentials(bearer=bearer, cookie=None)
+        return await self._get_credentials(bearer=bearer, cookie=None)
 
 
 class JwtAccessCookie(JwtAccess):
@@ -308,11 +319,11 @@ def __init__(
             refresh_expires_delta=refresh_expires_delta,
         )
 
-    def __call__(
+    async def __call__(
         self,
         cookie: JwtAuthBase.JwtAccessCookie = Security(JwtAccess._cookie),
     ) -> Optional[JwtAuthorizationCredentials]:
-        return self._get_credentials(bearer=None, cookie=cookie)
+        return await self._get_credentials(bearer=None, cookie=cookie)
 
 
 class JwtAccessBearerCookie(JwtAccess):
@@ -333,12 +344,12 @@ def __init__(
             refresh_expires_delta=refresh_expires_delta,
         )
 
-    def __call__(
+    async def __call__(
         self,
         bearer: JwtAuthBase.JwtAccessBearer = Security(JwtAccess._bearer),
         cookie: JwtAuthBase.JwtAccessCookie = Security(JwtAccess._cookie),
     ) -> Optional[JwtAuthorizationCredentials]:
-        return self._get_credentials(bearer=bearer, cookie=cookie)
+        return await self._get_credentials(bearer=bearer, cookie=cookie)
 
 
 class JwtRefresh(JwtAuthBase):
@@ -363,12 +374,12 @@ def __init__(
             refresh_expires_delta=refresh_expires_delta,
         )
 
-    def _get_credentials(
+    async def _get_credentials(
         self,
         bearer: Optional[JwtAuthBase.JwtRefreshBearer],
         cookie: Optional[JwtAuthBase.JwtRefreshCookie],
     ) -> Optional[JwtAuthorizationCredentials]:
-        payload = self._get_payload(bearer, cookie)
+        payload = await self._get_payload(bearer, cookie)
 
         if payload is None:
             return None
@@ -405,10 +416,10 @@ def __init__(
             refresh_expires_delta=refresh_expires_delta,
         )
 
-    def __call__(
+    async def __call__(
         self, bearer: JwtAuthBase.JwtRefreshBearer = Security(JwtRefresh._bearer)
     ) -> Optional[JwtAuthorizationCredentials]:
-        return self._get_credentials(bearer=bearer, cookie=None)
+        return await self._get_credentials(bearer=bearer, cookie=None)
 
 
 class JwtRefreshCookie(JwtRefresh):
@@ -429,11 +440,11 @@ def __init__(
             refresh_expires_delta=refresh_expires_delta,
         )
 
-    def __call__(
+    async def __call__(
         self,
         cookie: JwtAuthBase.JwtRefreshCookie = Security(JwtRefresh._cookie),
     ) -> Optional[JwtAuthorizationCredentials]:
-        return self._get_credentials(bearer=None, cookie=cookie)
+        return await self._get_credentials(bearer=None, cookie=cookie)
 
 
 class JwtRefreshBearerCookie(JwtRefresh):
@@ -454,9 +465,9 @@ def __init__(
             refresh_expires_delta=refresh_expires_delta,
         )
 
-    def __call__(
+    async def __call__(
         self,
         bearer: JwtAuthBase.JwtRefreshBearer = Security(JwtRefresh._bearer),
         cookie: JwtAuthBase.JwtRefreshCookie = Security(JwtRefresh._cookie),
     ) -> Optional[JwtAuthorizationCredentials]:
-        return self._get_credentials(bearer=bearer, cookie=cookie)
+        return await self._get_credentials(bearer=bearer, cookie=cookie)

setup.cfg

@@ -96,7 +96,6 @@ allow_untyped_decorators = True
 [tool:pytest]
 testpaths = tests/
 python_files = test_*.py
-timeout = 180
 addopts = --cov=fastapi_jwt/ --cov-report term-missing
 
 [isort]

tests/test_security_jwt_general.py

@@ -1,5 +1,5 @@
 import datetime
-from uuid import uuid1
+from uuid import uuid4
 from typing import Set
 
 from fastapi import FastAPI, Security
@@ -11,7 +11,7 @@
 app = FastAPI()
 
 access_security = JwtAccessBearer(secret_key="secret_key")
-refresh_security = JwtRefreshBearer(secret_key="secret_key")
+refresh_security = JwtRefreshBearer.from_other(access_security)
 
 
 unique_identifiers_database: Set[str] = set()
@@ -20,7 +20,7 @@
 @app.post("/auth")
 def auth():
     subject = {"username": "username", "role": "user"}
-    unique_identifier = str(uuid1())
+    unique_identifier = str(uuid4())
     unique_identifiers_database.add(unique_identifier)
 
     access_token = access_security.create_access_token(
@@ -33,7 +33,7 @@ def auth():
 
 @app.post("/refresh")
 def refresh(credentials: JwtAuthorizationCredentials = Security(refresh_security)):
-    unique_identifier = str(uuid1())
+    unique_identifier = str(uuid4())
     unique_identifiers_database.add(unique_identifier)
 
     access_token = refresh_security.create_access_token(

tests/test_security_jwt_general_optional.py

@@ -1,5 +1,5 @@
 import datetime
-from uuid import uuid1
+from uuid import uuid4
 from typing import Optional, Set
 
 from fastapi import FastAPI, Security
@@ -11,7 +11,7 @@
 app = FastAPI()
 
 access_security = JwtAccessBearer(secret_key="secret_key", auto_error=False)
-refresh_security = JwtRefreshBearer(secret_key="secret_key", auto_error=False)
+refresh_security = JwtRefreshBearer.from_other(access_security)
 
 
 unique_identifiers_database: Set[str] = set()
@@ -20,7 +20,7 @@
 @app.post("/auth")
 def auth():
     subject = {"username": "username", "role": "user"}
-    unique_identifier = str(uuid1())
+    unique_identifier = str(uuid4())
     unique_identifiers_database.add(unique_identifier)
 
     access_token = access_security.create_access_token(
@@ -38,7 +38,7 @@ def refresh(
     if credentials is None:
         return {"msg": "Create an account first"}
 
-    unique_identifier = str(uuid1())
+    unique_identifier = str(uuid4())
     unique_identifiers_database.add(unique_identifier)
 
     access_token = refresh_security.create_access_token(