SignUp and SignIn process

api/dependensies/authorizations.py

@@ -0,0 +1,52 @@
+from datetime import datetime
+from typing import Annotated
+
+from fastapi import Depends, HTTPException
+from fastapi.security import OAuth2PasswordBearer
+from jose import jwt
+from pydantic import BaseModel, ValidationError
+from starlette import status
+
+from api.error_codes import ErrorCode
+from api.services.security import JwtService
+from core import settings
+from db.models import User
+from db.queries.user import UserQueryService
+from db.session import SessionDep
+
+reusable_oauth = OAuth2PasswordBearer(
+    tokenUrl="/api/v1/auth/sign-in",
+    # scheme_name="JWT"
+)
+TokenDep = Annotated[str, Depends(reusable_oauth)]
+
+
+class TokenPayload(BaseModel):
+    sub: int
+    exp: int
+
+
+async def get_current_user(session: SessionDep, token: TokenDep) -> User:
+    try:
+        payload = JwtService().decode_token(token)
+        token_data = TokenPayload(**payload)
+    except (jwt.JWTError, ValidationError):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=ErrorCode.TOKEN_INVALID.to_json(),
+        )
+    if datetime.fromtimestamp(token_data.exp) < datetime.now():
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=ErrorCode.TOKEN_EXPIRED.to_json(),
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    user = await UserQueryService(session).get_user_by_id(token_data.sub)
+    if not user:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ErrorCode.USER_NOT_FOUND.to_json())
+    if not user.is_active:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ErrorCode.NOT_ACTIVE.to_json())
+    return user
+
+
+CurrentUser = Annotated[User, Depends(get_current_user)]

api/error_codes.py

@@ -1,18 +1,30 @@
 from enum import Enum
+from typing import TypedDict
+
+
+class ErrorCodeT(TypedDict):
+    code: str
+    detail: str
 
 
 class ErrorCode(Enum):
     WRONG_CREDENTIALS = ('user.auth.wrong_credentials', 'Email or password is incorrect')
     NOT_ACTIVE = ('user.auth.is_not_active', 'User is not active')
+    USER_NOT_FOUND = ('user.not_found', 'User not found')
+    TOKEN_EXPIRED = ('token.expired', 'Token expired')
+    TOKEN_INVALID = ('token.invalid', 'Could not validate credentials')
 
     def __init__(self, code: str, detail: str):
         self._code = code
         self._detail = detail
 
     @property
-    def code(self):
+    def code(self) -> str:
         return self._code
 
     @property
-    def detail(self):
+    def detail(self) -> str:
         return self._detail
+
+    def to_json(self) -> ErrorCodeT:
+        return {"code": self.code, "detail": self.detail}

api/services/handlers.py

@@ -5,6 +5,4 @@
 
 
 async def validation_exception_handler(request: Request, exc: ValidationError) -> JSONResponse:
-    return JSONResponse(status_code=400, content={"detail": {
-        "code": exc.code, "detail": exc.detail
-    }})
+    return JSONResponse(status_code=400, content={"detail": {"code": exc.code, "detail": exc.detail}})

api/services/security.py

@@ -1,11 +1,16 @@
 from datetime import datetime, timedelta
 from typing import Any
 
+from fastapi.security import OAuth2PasswordBearer
 from jose import jwt
 from passlib.context import CryptContext
 
 from core import settings
 
+reusable_oauth2 = OAuth2PasswordBearer(
+    tokenUrl=f"/api/v1/auth/sign-in",
+)
+
 
 class JwtService:
 
@@ -21,6 +26,11 @@ def __init__(
         self.access_token_secret_key = settings.jwt.JWT_SECRET_KEY
         self.refresh_token_secret_key = settings.jwt.JWT_REFRESH_SECRET_KEY
 
+    def get_tokens(self, user_id: int) -> dict:
+        access_token = self.create_access_token(subject=user_id)
+        refresh_token = self.create_refresh_token(subject=user_id)
+        return {"access_token": access_token, "refresh_token": refresh_token}
+
     def create_access_token(self, subject: str | Any) -> str:
         return self.__get_encoded_jwt(subject, self.access_token_expire, self.access_token_secret_key)
 
@@ -37,6 +47,10 @@ def __get_encoded_jwt(self, subject: str, expires_minutes: int, secret_key: str)
         )
         return encoded_jwt
 
+    def decode_token(self, token: str) -> dict:
+        payload = jwt.decode(token, self.access_token_secret_key, algorithms=[self.algorithm])
+        return payload
+
 
 class PasswordManager:
     password_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

api/v1/auth/routers.py

@@ -16,3 +16,8 @@ def __init__(self):
             endpoint=views.sign_in_view,
             methods=['POST'],
         )
+        self._router.add_api_route(
+            path='/logout',
+            endpoint=views.logout_view,
+            methods=['POST'],
+        )

api/v1/auth/schemas.py

@@ -8,14 +8,16 @@ class UserSignUpSchema(BaseModel):
     password: str = Field(..., min_length=8, description="User password")
     password_confirm: str = Field(..., min_length=8, description="Password confirmation", exclude=True)
 
-    # class Config:
-    #     orm_mode = True
-
     @property
     def full_name(self) -> str:
         return f"{self.first_name} {self.last_name}"
 
 
-class UserSignInSchema(BaseModel):
-    email: EmailStr = Field(..., description="User email")
-    password: str = Field(..., min_length=8, description="User password")
+class TokenResponse(BaseModel):
+    access_token: str
+    refresh_token: str
+    token_type: str = "bearer"
+
+
+class SignUpResponse(BaseModel):
+    id: int

api/v1/auth/services.py

@@ -2,12 +2,13 @@
 from collections import defaultdict
 
 from fastapi import HTTPException
+from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.error_codes import ErrorCode
 from api.exceptions import ValidationError
 from api.services.security import JwtService, PasswordManager
-from api.v1.auth.schemas import UserSignInSchema, UserSignUpSchema
+from api.v1.auth.schemas import TokenResponse, UserSignInSchema, UserSignUpSchema
 from db.models import User
 from db.queries.user import UserQueryService
 
@@ -27,14 +28,14 @@ async def validate_password(self) -> defaultdict[str, list]:
         password = self.data.password
         if len(password) < 8:
             errors['password'].append("Password must be at least 8 characters long")
-        if not re.search(r'[A-Z]', password):
-            errors['password'].append("Password must contain at least one uppercase letter")
-        if not re.search(r'[a-z]', password):
-            errors['password'].append("Password must contain at least one lowercase letter")
-        if not re.search(r'[0-9]', password):
-            errors['password'].append("Password must contain at least one digit")
-        if not re.search(r'[^a-zA-Z0-9]', password):
-            errors['password'].append("Password must contain at least one special character")
+        # if not re.search(r'[A-Z]', password):
+        #     errors['password'].append("Password must contain at least one uppercase letter")
+        # if not re.search(r'[a-z]', password):
+        #     errors['password'].append("Password must contain at least one lowercase letter")
+        # if not re.search(r'[0-9]', password):
+        #     errors['password'].append("Password must contain at least one digit")
+        # if not re.search(r'[^a-zA-Z0-9]', password):
+        #     errors['password'].append("Password must contain at least one special character")
         if self.data.password != self.data.password_confirm:
             errors['password_confirm'].append("Passwords do not match")
         return errors
@@ -60,12 +61,12 @@ async def create_user(self) -> User:
 class LoginService:
     password_manager = PasswordManager
 
-    def __init__(self, session: AsyncSession, data: UserSignInSchema):
+    def __init__(self, session: AsyncSession, data: OAuth2PasswordRequestForm):
         self.data = data
         self.query = UserQueryService(session)
 
     async def authenticate(self) -> User:
-        user = await self.query.get_user_by_email(self.data.email)
+        user = await self.query.get_user_by_email(self.data.username)
         if not user:
             raise ValidationError(ErrorCode.WRONG_CREDENTIALS)
         if not self.password_manager().verify_password(self.data.password, user.password):
@@ -74,11 +75,11 @@ async def authenticate(self) -> User:
             raise ValidationError(ErrorCode.NOT_ACTIVE)
         return user
 
-    async def generate_response(self, user: User) -> dict:
+    async def generate_response(self, user: User) -> TokenResponse:
         service = JwtService()
         access_token = service.create_access_token(user.id)
         refresh_token = service.create_refresh_token(user.id)
-        return {
-            'access_token': access_token,
-            'refresh_token': refresh_token,
-        }
+        return TokenResponse(
+            access_token=access_token,
+            refresh_token=refresh_token,
+        )

api/v1/auth/views.py

@@ -1,17 +1,33 @@
+from typing import Annotated
+
+from fastapi import Depends
+from fastapi.security import OAuth2PasswordRequestForm
+
+from api.dependensies.authorizations import CurrentUser
 from db.session import SessionDep
 
-from .schemas import UserSignInSchema, UserSignUpSchema
+from .schemas import SignUpResponse, TokenResponse, UserSignUpSchema
 from .services import LoginService, SignUpService
 
 
-async def sign_up_view(data: UserSignUpSchema, session: SessionDep):
+async def sign_up_view(data: UserSignUpSchema, session: SessionDep) -> SignUpResponse:
     service = SignUpService(session, data)
     await service.validate()
     user = await service.create_user()
-    return {'detail': user.id}
+    return SignUpResponse(id=user.id)
 
 
-async def sign_in_view(data: UserSignInSchema, session: SessionDep):
-    service = LoginService(session, data)
+async def sign_in_view(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()], session: SessionDep
+) -> TokenResponse:
+    service = LoginService(session, form_data)
     user = await service.authenticate()
     return await service.generate_response(user)
+
+
+async def logout_view(current_user: CurrentUser):
+    print(f'{current_user=}')
+
+
+# user2@example.com
+# stringst

pyproject.toml

@@ -17,6 +17,7 @@ python-jose = "~=3.3"
 cryptography = "~=42.0"
 passlib = { version="~=1.7", extras = ["bcrypt"] }
 pydantic = { version="~=2.6", extras = ["email"] }
+python-multipart = "^0.0.9"
 
 
 [tool.poetry.group.dev]