Login and logout API

.github/workflows/tests.yml

@@ -35,7 +35,7 @@ jobs:
       - name: Install dependencies
         run: |
           pip install poetry
-          poetry install --with dev --no-interaction --no-ansi
+          poetry install --extras dev --no-interaction --no-ansi
       - name: Run pytest
         run: pytest
 

api/errors/__init__.py

@@ -0,0 +1,13 @@
+from .codes import ErrorCode
+from .exceptions import AppException, Forbidden, NotFound, Unauthorized, ValidationError
+from .handlers import app_exception_handler
+
+__all__ = [
+    "ErrorCode",
+    "AppException",
+    "Unauthorized",
+    "Forbidden",
+    "NotFound",
+    "ValidationError",
+    "app_exception_handler",
+]

api/error_codes.py → api/errors/codes.py

@@ -1,18 +1,27 @@
 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

api/errors/exceptions.py

@@ -0,0 +1,30 @@
+from fastapi import status
+
+from .codes import ErrorCode
+
+
+class AppException(Exception):
+    status_code: int
+
+    def __init__(self, error_code: ErrorCode):
+        self.code = error_code.code
+        self.detail = error_code.detail
+
+    def to_dict(self) -> dict:
+        return {"code": self.code, "detail": self.detail}
+
+
+class Unauthorized(AppException):
+    status_code = status.HTTP_401_UNAUTHORIZED
+
+
+class Forbidden(AppException):
+    status_code = status.HTTP_403_FORBIDDEN
+
+
+class NotFound(AppException):
+    status_code = status.HTTP_404_NOT_FOUND
+
+
+class ValidationError(AppException):
+    status_code = status.HTTP_400_BAD_REQUEST

api/errors/handlers.py

@@ -0,0 +1,8 @@
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+
+from api.errors.exceptions import AppException
+
+
+def app_exception_handler(request: Request, exc: AppException) -> JSONResponse:
+    return JSONResponse(status_code=exc.status_code, content=exc.to_dict())

api/services/dependencies/__init__.py

@@ -0,0 +1,4 @@
+from .authorization import CurrentUser
+from .session import SessionDep
+
+__all__ = ["SessionDep", "CurrentUser"]

api/services/dependencies/authorization.py

@@ -0,0 +1,38 @@
+from datetime import datetime
+from typing import Annotated
+
+from fastapi import Depends
+from fastapi.security import OAuth2PasswordBearer
+from jose import JWTError
+from pydantic import ValidationError
+
+from api.errors import ErrorCode, Forbidden, NotFound
+from api.services.security import JwtService
+from db.models import User
+from db.queries.user import UserQueryService
+
+from .session import SessionDep
+
+reusable_oauth = OAuth2PasswordBearer(
+    tokenUrl="/api/v1/auth/sign-in",
+)
+
+TokenDep = Annotated[str, Depends(reusable_oauth)]
+
+
+async def get_current_user(session: SessionDep, token: TokenDep) -> User:
+    try:
+        payload = JwtService().decode_token(token)
+    except (JWTError, ValidationError):
+        raise Forbidden(ErrorCode.TOKEN_INVALID)
+    if datetime.fromtimestamp(payload.exp) < datetime.now():
+        raise Forbidden(ErrorCode.TOKEN_EXPIRED)
+    user = await UserQueryService(session).get_user_by_id(payload.sub)
+    if not user:
+        raise NotFound(ErrorCode.USER_NOT_FOUND)
+    if not user.is_active:
+        raise NotFound(ErrorCode.NOT_ACTIVE)
+    return user
+
+
+CurrentUser = Annotated[User, Depends(get_current_user)]

api/services/dto.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class CreateResponseDTO(BaseModel):
+    id: int

api/services/security.py

@@ -3,10 +3,16 @@
 
 from jose import jwt
 from passlib.context import CryptContext
+from pydantic import BaseModel
 
 from core import settings
 
 
+class TokenPayload(BaseModel):
+    sub: int
+    exp: int
+
+
 class JwtService:
 
     def __init__(
@@ -21,6 +27,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 +48,10 @@ def __get_encoded_jwt(self, subject: str, expires_minutes: int, secret_key: str)
         )
         return encoded_jwt
 
+    def decode_token(self, token: str) -> TokenPayload:
+        payload = jwt.decode(token, self.access_token_secret_key, algorithms=[self.algorithm])
+        return TokenPayload.model_validate(payload)
+
 
 class PasswordManager:
     password_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

api/v1/auth/services.py

@@ -5,8 +5,7 @@
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from api.error_codes import ErrorCode
-from api.exceptions import ValidationError
+from api.errors import ErrorCode, ValidationError
 from api.services.security import JwtService, PasswordManager
 from api.v1.auth.schemas import TokenResponse, UserSignUpSchema
 from db.models import User

main.py

@@ -1,11 +1,9 @@
 import logging
 
 from fastapi import FastAPI
-from fastapi.routing import APIRoute
 from starlette.middleware.cors import CORSMiddleware
 
-from api.exceptions import ValidationError
-from api.services.handlers import validation_exception_handler
+from api.errors import AppException, app_exception_handler
 from api.v1.routers import router as v1_router
 from core import lifespan, settings
 
@@ -23,7 +21,7 @@ def get_application() -> FastAPI:
         lifespan=lifespan,
     )
     application.include_router(v1_router, prefix='/api/v1')
-    application.add_exception_handler(ValidationError, validation_exception_handler)
+    application.add_exception_handler(AppException, app_exception_handler)
     application.add_middleware(
         CORSMiddleware,
         allow_origins=settings.cors.allow_origins,