Замена libs и создание класса PWD в Security

Удалил passlib, теперь использую bcrypt и специально для него создал
класс PWD

Заменил python-jose на pyjwt

Author: ProFastCode

app/__init__.py

@@ -12,9 +12,9 @@
     root_path=settings.APP_PATH,
     version=settings.APP_VERSION,
     contact={
-        "name": "Fast Code",
-        "url": "https://fast-code.pro/",
-        "email": "fast.code.auth@gmail.com",
+        'name': 'Fast Code',
+        'url': 'https://fast-code.pro/',
+        'email': 'fast.code.auth@gmail.com',
     },
 )
 

app/api/endpoints/users/auth.py

@@ -1,24 +1,24 @@
 from fastapi import APIRouter
 
 from app.api import anotations
-from app.core import deps, exps
+from app.core import exps
 from app.models.token import AccessToken
 from app.models.user import UserCreate
 
-router = APIRouter(prefix="/auth")
+router = APIRouter(prefix='/auth')
 
 
-@router.post("/token/", response_model=AccessToken)
+@router.post('/token/', response_model=AccessToken)
 async def token(
     data: UserCreate, db: anotations.Database, security: anotations.Security
 ):
     """
     Retrieve new access token
     """
     if user := await db.user.retrieve_by_email(data.email):
-        if not deps.pwd_context.verify(data.password, user.password):
+        if not security.pwd.checkpwd(data.password, user.password):
             raise exps.USER_IS_CORRECT
-        access_token = security.jwt.encode_token({"id": user.id}, 1440)
+        access_token = security.jwt.encode_token({'id': user.id}, 1440)
         return AccessToken(token=access_token)
 
     raise exps.USER_NOT_FOUND

app/api/endpoints/users/create.py

@@ -1,24 +1,23 @@
 from fastapi import APIRouter
 
 from app.api import anotations
-from app.core import deps, exps
+from app.core import exps
 from app.models.user import User, UserCreate, UserRead
 
-router = APIRouter(prefix="/create")
+router = APIRouter(prefix='/create')
 
 
-@router.post("/", response_model=UserRead)
+@router.post('/', response_model=UserRead)
 async def registration(
-    data: UserCreate,
-    db: anotations.Database,
+    data: UserCreate, db: anotations.Database, security: anotations.Security
 ):
     """
     Create user
     """
     if await db.user.retrieve_by_email(data.email):
         raise exps.USER_EXISTS
 
-    password_hash = deps.pwd_context.hash(data.password)
+    password_hash = security.pwd.hashpwd(data.password)
     model = User(email=data.email, password=password_hash)
     user = await db.user.create(model)
     return user

app/core/__init__.py

@@ -4,4 +4,4 @@
 
 from .security import Security
 
-__all__ = ["Security"]
+__all__ = ['Security']

app/core/db.py

@@ -2,7 +2,8 @@
 Database
 """
 
-from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import (AsyncEngine, async_sessionmaker,
+                                    create_async_engine)
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app import repositories as repos

app/core/deps.py

@@ -4,15 +4,12 @@
 
 from fastapi import Depends
 from fastapi.security import APIKeyHeader
-from passlib.context import CryptContext
 from typing_extensions import Annotated, AsyncGenerator
 
 from app.models.user import User
 
 from . import Security, db, exps
 
-pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
-
 
 async def get_db() -> AsyncGenerator[db.Database]:
     async with db.SessionLocal() as session:

app/core/exps.py

@@ -8,7 +8,7 @@
 USER_EXISTS = HTTPException(status.HTTP_409_CONFLICT, 'User is already taken.')
 USER_NOT_FOUND = HTTPException(status.HTTP_404_NOT_FOUND, 'User not found.')
 USER_IS_CORRECT = HTTPException(
-    status.HTTP_401_UNAUTHORIZED, 'Authorization failed. Please try again'
+    status.HTTP_401_UNAUTHORIZED, 'User is correct'
 )
 
 # Tokens

app/core/security/__init__.py

@@ -1,11 +1,13 @@
 from app.core.settings import settings
 
 from .jwt import JWT
+from .pwd import PWD
 
 
 class Security:
     def __init__(self):
         self.jwt = JWT(settings.APP_SECRET_KEY)
+        self.pwd = PWD()
 
 
-__all__ = ["Security"]
+__all__ = ['Security']

app/core/security/jwt.py

@@ -1,7 +1,7 @@
 import datetime as dt
 
-from jose import JWTError, jwt
-from jose.constants import ALGORITHMS
+import jwt
+from jwt import PyJWTError
 
 from app.core import exps
 
@@ -12,10 +12,8 @@ def __init__(self, secret_key: str):
 
     def decode_token(self, token: str) -> dict | None:
         try:
-            payload = jwt.decode(
-                token, self.secret_key, algorithms=[ALGORITHMS.HS256]
-            )
-        except JWTError:
+            payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
+        except PyJWTError:
             raise exps.TOKEN_INVALID
 
         exp = payload.get('exp')
@@ -28,4 +26,4 @@ def encode_token(self, payload: dict, minutes: int) -> str:
             'payload': payload,
             'exp': dt.datetime.now(dt.UTC) + dt.timedelta(minutes=minutes),
         }
-        return jwt.encode(claims, self.secret_key, algorithm=ALGORITHMS.HS256)
+        return jwt.encode(claims, self.secret_key, algorithm='HS256')

app/core/security/pwd.py

@@ -0,0 +1,9 @@
+import bcrypt
+
+
+class PWD:
+    def hashpwd(self, password: str) -> str:
+        return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
+
+    def checkpwd(self, password: str, hashed_password: str) -> bool:
+        return bcrypt.checkpw(password.encode(), hashed_password.encode())