add mixins and jwt strategy

Author: sumaro2101

api_v1/auth/__init__.py

@@ -0,0 +1,14 @@
+from fastapi_users.authentication import (
+    BearerTransport,
+    AuthenticationBackend,
+    JWTStrategy,
+    )
+
+from config import settings
+
+
+bearer_transport = BearerTransport()
+
+
+def get_jwt_strategy() -> JWTStrategy:
+    return JWTStrategy(secret=settings.SECRET)

api_v1/auth/backends.py

@@ -0,0 +1,11 @@
+from fastapi import APIRouter
+
+
+router = APIRouter(prefix='/auth',
+                   tags=['Auth'],
+                   )
+
+
+
+@router.post(path='/login',
+            )

api_v1/auth/schemas.py

@@ -2,6 +2,7 @@
 
 from config import settings
 from api_v1.users.views import router as users
+from api_v1.auth.views import router as auth
 
 
 def register_routers(app: FastAPI) -> None:
@@ -13,7 +14,7 @@ def register_routers(app: FastAPI) -> None:
 
     ## Returns:
         None
-    
+
     ## Example
     ```python
     from fastapi import FastAPI
@@ -39,3 +40,8 @@ def register_routers(app: FastAPI) -> None:
         router=users,
         prefix=settings.API_PREFIX,
         )
+
+    app.include_router(
+        router=auth,
+        prefix=settings.API_PREFIX,
+        )

api_v1/auth/views.py

@@ -0,0 +1,57 @@
+import jwt
+from fastapi_users.jwt import decode_jwt
+from fastapi_users import exceptions
+
+from .exceptions import UserNotVerified, PasswordNotValidError
+
+
+class AuthenticationUserManagerMixin:
+    """
+    Миксин добавляющий аутентификацию в логику менеджера
+    """
+
+    async def authenticate_user(self, token: str | None):
+        if not token:
+            return
+        try:
+            data = decode_jwt(
+                token,
+                self.verification_token_secret,
+                [self.verification_token_audience],
+            )
+        except jwt.PyJWTError:
+            raise exceptions.InvalidVerifyToken()
+
+        try:
+            user_id = data["sub"]
+            email = data["email"]
+        except KeyError:
+            raise exceptions.InvalidVerifyToken()
+
+        try:
+            user = await self.get_by_email(email)
+        except exceptions.UserNotExists:
+            return
+
+        try:
+            parsed_id = self.parse_id(user_id)
+        except exceptions.InvalidID:
+            raise exceptions.InvalidVerifyToken()
+
+        if parsed_id != user.id:
+            raise exceptions.InvalidVerifyToken()
+
+        if not user.is_verified:
+            raise UserNotVerified()
+
+        return user
+
+
+class PasswordValidationMixin:
+    """
+    Миксин добавляющий валидацию пароля
+    """
+
+    async def validate_password(self, password, user):
+        if not len(password) > 7:
+            raise PasswordNotValidError('Password is not valid')

api_v1/routers.py

@@ -1,56 +1,17 @@
-import jwt
 from fastapi_users import BaseUserManager, IntegerIDMixin
-from fastapi_users.jwt import decode_jwt
-from fastapi_users import exceptions
 
-from .exceptions import PasswordNotValidError, UserNotVerified
+from .mixins import AuthenticationUserManagerMixin, PasswordValidationMixin
 from config.models import User
 from config import settings
 
 
-class UserManager(IntegerIDMixin, BaseUserManager[User, int]):
+class UserManager(AuthenticationUserManagerMixin,
+                  PasswordValidationMixin,
+                  IntegerIDMixin,
+                  BaseUserManager[User, int]):
     """
     UserManager для работы с пользователем
     """
 
     verification_token_secret = settings.SECRET
-
-    async def validate_password(self, password, user):
-        if not len(password) > 7:
-            raise PasswordNotValidError('Password is not valid')
-
-    async def authenticate_user(self, token: str | None):
-        if not token:
-            return
-        try:
-            data = decode_jwt(
-                token,
-                self.verification_token_secret,
-                [self.verification_token_audience],
-            )
-        except jwt.PyJWTError:
-            raise exceptions.InvalidVerifyToken()
-
-        try:
-            user_id = data["sub"]
-            email = data["email"]
-        except KeyError:
-            raise exceptions.InvalidVerifyToken()
-
-        try:
-            user = await self.get_by_email(email)
-        except exceptions.UserNotExists:
-            return
-
-        try:
-            parsed_id = self.parse_id(user_id)
-        except exceptions.InvalidID:
-            raise exceptions.InvalidVerifyToken()
-
-        if parsed_id != user.id:
-            raise exceptions.InvalidVerifyToken()
-
-        if not user.is_verified:
-            raise UserNotVerified()
-
-        return user
+    reset_password_token_lifetime_seconds = settings.RESET_LIFESPAN_TOKEN_SECONDS

api_v1/users/mixins.py

@@ -1,8 +1,4 @@
-from fastapi import APIRouter, Depends, status, Request
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from config import db_connection
-from api_v1.exeptions import ValidationError
+from fastapi import APIRouter, Request
 
 
 router = APIRouter(prefix='/users',

api_v1/users/user_manager.py

@@ -97,6 +97,7 @@ class Settings(BaseSettings):
     LOG_DIR: Path = log_dir
     CURRENT_ORIGIN: str = config('CURRENT_ORIGIN')
     SECRET: str = config('SECRET')
+    RESET_LIFESPAN_TOKEN_SECONDS: int = 3600
 
 
 settings = Settings()