Extract token refresh service

Author: koldakov

futuramaapi/routers/rest/tokens/api.py

@@ -9,9 +9,11 @@
     GetAuthUserTokenResponse,
     GetAuthUserTokenService,
 )
-
-from .dependencies import refresh_token
-from .schemas import UserToken
+from futuramaapi.routers.services.tokens.get_refreshed_auth_user_token import (
+    GetRefreshedAuthUserTokenRequest,
+    GetRefreshedAuthUserTokenResponse,
+    GetRefreshedAuthUserTokenService,
+)
 
 router: APIRouter = APIRouter(
     prefix="/tokens",
@@ -57,15 +59,18 @@ async def get_user_auth_token(
             "model": UnauthorizedResponse,
         },
     },
-    response_model=UserToken,
-    name="user_token_auth_refresh",
+    response_model=GetRefreshedAuthUserTokenResponse,
+    name="get_refreshed_user_auth_token",
 )
-async def refresh_token_auth_user(
-    token: Annotated[UserToken, Depends(refresh_token)],
-) -> UserToken:
+async def get_refreshed_user_auth_token(
+    data: GetRefreshedAuthUserTokenRequest,
+) -> GetRefreshedAuthUserTokenResponse:
     """Refresh JWT.
 
     The Refresh JWT Token endpoint extends the lifespan of JSON Web Tokens (JWTs) without requiring user
     reauthentication. This API feature ensures uninterrupted access to secured resources.
     """
-    return token
+    service: GetRefreshedAuthUserTokenService = GetRefreshedAuthUserTokenService(
+        request_data=data,
+    )
+    return await service()

futuramaapi/routers/rest/tokens/dependencies.py

@@ -1,6 +1,4 @@
-from typing import TYPE_CHECKING, ClassVar, Literal, Self
-
-from pydantic import Field
+from typing import TYPE_CHECKING, Literal, Self
 
 from futuramaapi.helpers.pydantic import BaseModel, BaseTokenModel
 from futuramaapi.mixins.pydantic import BaseModelTokenMixin, DecodedTokenError
@@ -37,51 +35,3 @@ def decode(
             raise DecodedTokenError() from None
 
         return decoded_token
-
-
-class UserTokenRefreshRequest(BaseModel):
-    refresh_token: str
-
-
-class UserToken(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(
-        cls,
-        user: "User",
-        /,
-    ) -> Self:
-        access: DecodedUserToken = DecodedUserToken.from_user(user, "access")
-        refresh: DecodedUserToken = DecodedUserToken.from_user(user, "refresh")
-        return cls(
-            access_token=access.tokenize(cls._default_access_seconds),
-            refresh_token=refresh.tokenize(cls._default_refresh_seconds),
-        )
-
-    def refresh(self) -> None:
-        try:
-            access: DecodedUserToken = DecodedUserToken.decode(self.access_token)
-        except DecodedTokenError:
-            raise
-
-        try:
-            refresh: DecodedUserToken = DecodedUserToken.decode(self.refresh_token, allowed_type="refresh")
-        except DecodedTokenError:
-            raise
-
-        access.refresh_nonce()
-        refresh.refresh_nonce()
-
-        self.access_token = access.tokenize(self._default_access_seconds)
-        self.refresh_token = refresh.tokenize(self._default_refresh_seconds)

futuramaapi/routers/rest/tokens/schemas.py

@@ -0,0 +1,60 @@
+from typing import Any, ClassVar
+
+import jwt
+from fastapi import HTTPException, status
+from jwt import ExpiredSignatureError, InvalidSignatureError, InvalidTokenError
+from pydantic import Field
+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
+
+from .get_auth_user_token import GetAuthUserTokenResponse
+
+
+class GetRefreshedAuthUserTokenRequest(BaseModel):
+    refresh_token: str = Field(
+        alias="refresh_token",
+    )
+
+
+class GetRefreshedAuthUserTokenResponse(GetAuthUserTokenResponse):
+    pass
+
+
+class GetRefreshedAuthUserTokenService(BaseSessionService[GetRefreshedAuthUserTokenResponse]):
+    request_data: GetRefreshedAuthUserTokenRequest
+
+    algorithm: ClassVar[str] = "HS256"
+
+    def __get_user_statement(self, decoded_token: dict[str, Any], /) -> Select[tuple[UserModel]]:
+        return select(UserModel).where(UserModel.id == decoded_token["user"]["id"])
+
+    async def _get_user(self, decoded_token: dict[str, Any], /) -> UserModel:
+        result: Result[tuple[UserModel]] = await self.session.execute(self.__get_user_statement(decoded_token))
+        try:
+            user: UserModel = result.scalars().one()
+        except NoResultFound:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED) from None
+
+        return user
+
+    async def process(self, *args, **kwargs) -> GetRefreshedAuthUserTokenResponse:
+        try:
+            decoded_token: dict[str, Any] = jwt.decode(
+                self.request_data.refresh_token,
+                key=settings.secret_key.get_secret_value(),
+                algorithms=[self.algorithm],
+            )
+        except (ExpiredSignatureError, InvalidSignatureError, InvalidTokenError):
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED) from None
+
+        if decoded_token["type"] != "refresh":
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+
+        user: UserModel = await self._get_user(decoded_token)
+
+        return GetRefreshedAuthUserTokenResponse.from_user_model(user)