Extract get user auth token service

Author: koldakov

futuramaapi/routers/rest/tokens/api.py

@@ -1,11 +1,16 @@
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, status
+from fastapi.security import OAuth2PasswordRequestForm
+from pydantic import SecretStr
 
 from futuramaapi.routers.exceptions import UnauthorizedResponse
-from futuramaapi.routers.rest.users.schemas import User
+from futuramaapi.routers.services.tokens.get_auth_user_token import (
+    GetAuthUserTokenResponse,
+    GetAuthUserTokenService,
+)
 
-from .dependencies import from_form_data, refresh_token
+from .dependencies import refresh_token
 from .schemas import UserToken
 
 router: APIRouter = APIRouter(
@@ -21,12 +26,15 @@
             "model": UnauthorizedResponse,
         },
     },
-    response_model=UserToken,
-    name="user_token_auth",
+    response_model=GetAuthUserTokenResponse,
+    name="get_user_auth_token",
 )
-async def token_auth_user(
-    user: Annotated[User, Depends(from_form_data)],
-) -> UserToken:
+async def get_user_auth_token(
+    form_data: Annotated[
+        OAuth2PasswordRequestForm,
+        Depends(),
+    ],
+) -> GetAuthUserTokenResponse:
     """Authenticate user.
 
     JSON Web Token (JWT) authentication is a popular method for securing web applications and APIs.
@@ -35,7 +43,11 @@ async def token_auth_user(
 
     Use a token in a response to get secured stored data of your user.
     """
-    return UserToken.from_user(user)
+    service: GetAuthUserTokenService = GetAuthUserTokenService(
+        username=form_data.username,
+        password=SecretStr(form_data.password),
+    )
+    return await service()
 
 
 @router.post(

futuramaapi/routers/rest/tokens/dependencies.py

@@ -1,30 +1,13 @@
 from typing import Annotated
 
 from fastapi import Depends, HTTPException, status
-from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
-from sqlalchemy.ext.asyncio.session import AsyncSession
-
-from futuramaapi.repositories.session import get_async_session
-from futuramaapi.routers.exceptions import ModelNotFoundError
-from futuramaapi.routers.rest.users.schemas import User, UserPasswordError
+from fastapi.security import OAuth2PasswordBearer
 
 from .schemas import DecodedTokenError, UserToken, UserTokenRefreshRequest
 
 _oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/tokens/users/auth")
 
 
-async def from_form_data(
-    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
-    session: AsyncSession = Depends(get_async_session),  # noqa: B008
-) -> User:
-    try:
-        user: User = await User.auth(session, form_data.username, form_data.password)
-    except (ModelNotFoundError, UserPasswordError):
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED) from None
-
-    return user
-
-
 def refresh_token(
     token: Annotated[str, Depends(_oauth2_scheme)],
     data: UserTokenRefreshRequest,

futuramaapi/routers/services/tokens/__init__.py

@@ -0,0 +1,88 @@
+import uuid
+from datetime import UTC, datetime, timedelta
+from typing import Any, ClassVar, Self
+
+import jwt
+from fastapi import HTTPException, status
+from pydantic import Field, SecretStr
+from sqlalchemy import Result, Select, select
+from sqlalchemy.exc import NoResultFound
+
+from futuramaapi.core import settings
+from futuramaapi.helpers.pydantic import BaseModel
+from futuramaapi.repositories.models import UserModel
+from futuramaapi.routers.services import BaseSessionService
+
+
+class GetAuthUserTokenResponse(BaseModel):
+    access_token: str = Field(
+        alias="access_token",
+        description="Keep in mind, that the field is not in a camel case. That's the standard.",
+    )
+    refresh_token: str = Field(
+        alias="refresh_token",
+        description="Keep in mind, that the field is not in a camel case. That's the standard.",
+    )
+
+    _default_access_seconds: ClassVar[int] = 15 * 60
+    _default_refresh_seconds: ClassVar[int] = 5 * 24 * 60 * 60
+
+    @classmethod
+    def from_user_model(
+        cls,
+        user: UserModel,
+        /,
+        *,
+        algorithm="HS256",
+    ) -> Self:
+        return cls(
+            access_token=jwt.encode(
+                {
+                    "exp": datetime.now(UTC) + timedelta(seconds=cls._default_access_seconds),
+                    "nonce": uuid.uuid4().hex,
+                    "type": "access",
+                    "user": {
+                        "id": user.id,
+                    },
+                },
+                settings.secret_key.get_secret_value(),
+                algorithm=algorithm,
+            ),
+            refresh_token=jwt.encode(
+                {
+                    "exp": datetime.now(UTC) + timedelta(seconds=cls._default_refresh_seconds),
+                    "nonce": uuid.uuid4().hex,
+                    "type": "refresh",
+                    "user": {
+                        "id": user.id,
+                    },
+                },
+                settings.secret_key.get_secret_value(),
+                algorithm=algorithm,
+            ),
+        )
+
+
+class GetAuthUserTokenService(BaseSessionService[GetAuthUserTokenResponse]):
+    username: str
+    password: SecretStr
+
+    @property
+    def __user_statement(self) -> Select[tuple[UserModel]]:
+        return select(UserModel).where(UserModel.username == self.username)
+
+    async def _get_user(self) -> UserModel:
+        result: Result[tuple[Any]] = await self.session.execute(self.__user_statement)
+        try:
+            return result.scalars().one()
+        except NoResultFound:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+            ) from None
+
+    async def process(self, *args, **kwargs) -> GetAuthUserTokenResponse:
+        user: UserModel = await self._get_user()
+        if not self.hasher.verify(self.password.get_secret_value(), user.password):
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+
+        return GetAuthUserTokenResponse.from_user_model(user)