small refactor

src/app/application/dto/auth.py

@@ -0,0 +1,36 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional
+
+from src.app.application.common.dto.base import AppBaseDTO
+
+
+@dataclass
+class TokenPairDTO(AppBaseDTO):
+    """DTO for token pair (access and refresh tokens)."""
+
+    access_token: str
+    refresh_token: str
+
+    def to_dict(self) -> dict:
+        return {
+            "access": self.access_token,
+            "refresh": self.refresh_token,
+        }
+
+
+@dataclass
+class DecodedTokenDTO(AppBaseDTO):
+    """DTO for decoded token data."""
+
+    uuid: str
+    sid: str
+    token_type: str
+    exp: Optional[datetime] = None
+
+    def to_dict(self) -> dict:
+        return {
+            "uuid": self.uuid,
+            "sid": self.sid,
+            "token_type": self.token_type,
+        }

src/app/application/dto/user.py

@@ -15,3 +15,19 @@ class UserShortDTO(AppBaseDTO):
     email: Optional[str]
     phone: Optional[str]
     is_active: bool
+
+
+@dataclass
+class CreateUserByEmailDTO(AppBaseDTO):
+    """DTO for creating user by email and password."""
+
+    email: str
+    password: str
+
+
+@dataclass
+class CreateUserByPhoneDTO(AppBaseDTO):
+    """DTO for creating user by phone and verification code."""
+
+    phone: str
+    verification_code: str

src/app/application/services/auth_service.py

@@ -4,9 +4,9 @@
 
 from src.app.application.common.services.base import AbstractBaseApplicationService
 from src.app.application.container import container as app_services_container, ApplicationServicesContainer
+from src.app.application.dto.auth import DecodedTokenDTO, TokenPairDTO
 from src.app.application.dto.user import UserShortDTO
 from src.app.domain.auth.container import container as domain_auth_svc_container, DomainAuthServiceContainer
-from src.app.domain.auth.value_objects import TokenPair, DecodedToken
 from src.app.domain.common.exceptions import ValidationError
 from src.app.domain.common.utils.common import mask_string
 from src.app.domain.users.container import container as domain_users_svc_container, DomainUsersServiceContainer
@@ -45,22 +45,38 @@ async def get_auth_user_by_email_password(cls, email: str, password: str) -> Any
         return user
 
     @classmethod
-    def verify_access_token(cls, token: str) -> DecodedToken:
+    def verify_access_token(cls, token: str) -> DecodedTokenDTO:
         """Verify an access token and return decoded data."""
-        return cls.dom_auth_svc_container.jwt_service.verify_access_token(token)
+        decoded_vo = cls.dom_auth_svc_container.jwt_service.verify_access_token(token)
+        return DecodedTokenDTO(
+            uuid=decoded_vo.uuid,
+            sid=decoded_vo.sid,
+            token_type=decoded_vo.token_type.value,
+            exp=decoded_vo.exp,
+        )
 
     @classmethod
-    def verify_refresh_token(cls, token: str) -> DecodedToken:
+    def verify_refresh_token(cls, token: str) -> DecodedTokenDTO:
         """Verify a refresh token and return decoded data."""
-        return cls.dom_auth_svc_container.jwt_service.verify_refresh_token(token)
+        decoded_vo = cls.dom_auth_svc_container.jwt_service.verify_refresh_token(token)
+        return DecodedTokenDTO(
+            uuid=decoded_vo.uuid,
+            sid=decoded_vo.sid,
+            token_type=decoded_vo.token_type.value,
+            exp=decoded_vo.exp,
+        )
 
     @classmethod
-    def create_tokens_for_user(cls, uuid: str) -> TokenPair:
+    def create_tokens_for_user(cls, uuid: str) -> TokenPairDTO:
         """Create access and refresh tokens for a user."""
-        return cls.dom_auth_svc_container.jwt_service.create_token_pair(uuid)
+        token_pair_vo = cls.dom_auth_svc_container.jwt_service.create_token_pair(uuid)
+        return TokenPairDTO(
+            access_token=token_pair_vo.access_token,
+            refresh_token=token_pair_vo.refresh_token,
+        )
 
     @classmethod
-    async def refresh_tokens(cls, refresh_token: str) -> tuple[UserShortDTO, TokenPair]:
+    async def refresh_tokens(cls, refresh_token: str) -> tuple[UserShortDTO, TokenPairDTO]:
         """Verify refresh token, get user, and create new token pair."""
         decoded = cls.verify_refresh_token(refresh_token)
         user = await cls.app_svc_container.users_service.get_first(

src/app/application/services/users_service.py

@@ -4,12 +4,12 @@
 
 from src.app.application.common.services.base import BaseApplicationService
 from src.app.application.container import container as app_services_container, ApplicationServicesContainer
-from src.app.application.dto.user import UserShortDTO
+from src.app.application.dto.user import CreateUserByEmailDTO, CreateUserByPhoneDTO, UserShortDTO
 from src.app.domain.auth.container import container as domain_auth_svc_container, DomainAuthServiceContainer
 from src.app.domain.common.exceptions import AlreadyExistsError
 from src.app.domain.common.utils.common import mask_string
 from src.app.domain.users.container import container as domain_users_svc_container, DomainUsersServiceContainer
-from src.app.domain.users.value_objects.users_vob import EmailPasswordPair, PhoneNumberCodePair
+from src.app.domain.users.value_objects.users_vo import EmailPasswordPair, PhoneNumberCodePair
 from src.app.infrastructure.repositories.container import container as repo_container
 
 
@@ -28,21 +28,28 @@ class AppUserService(BaseApplicationService):
     @classmethod
     async def create_user_by_email(cls, email: str, password: str) -> Optional[UserShortDTO]:
         _, validated_email = validate_email(email)
-        validated_data = EmailPasswordPair(email=validated_email, password=password)
-        password_hashed = domain_auth_svc_container.auth_service.get_password_hashed(password=password)
+        input_dto = CreateUserByEmailDTO(email=validated_email, password=password)
+
+        email_password_vo = EmailPasswordPair(email=input_dto.email, password=input_dto.password)
+
+        # Use domain service for password hashing
+        password_hashed = domain_auth_svc_container.auth_service.get_password_hashed(password=input_dto.password)
+
+        # Check business rule: email uniqueness
         is_email_exists = await cls.app_svc_container.users_service.is_exists(
-            filter_data={"email": validated_data.email}
+            filter_data={"email": email_password_vo.email}
         )
         if is_email_exists or not email:
             raise AlreadyExistsError(
                 message="Already exists",
                 details=[
-                    {"key": "email", "value": mask_string(validated_data.email, keep_start=1, keep_end=4)},
+                    {"key": "email", "value": mask_string(email_password_vo.email, keep_start=1, keep_end=4)},
                 ],
             )
 
+        # Prepare persistence data
         data = {
-            "email": validated_data.email,
+            "email": email_password_vo.email,
             "password_hashed": password_hashed,
         }
         user_dto = await cls.create(data, is_return_require=True, out_dataclass=UserShortDTO)
@@ -51,16 +58,28 @@ async def create_user_by_email(cls, email: str, password: str) -> Optional[UserS
 
     @classmethod
     async def create_user_by_phone(cls, phone: str, verification_code: str) -> Optional[UserShortDTO]:
-        validated_data = PhoneNumberCodePair(phone=phone, verification_code=verification_code)
+        # Create application DTO
+        input_dto = CreateUserByPhoneDTO(phone=phone, verification_code=verification_code)
+
+        # Convert to domain value object for validation
+        phone_code_vo = PhoneNumberCodePair(phone=input_dto.phone, verification_code=input_dto.verification_code)
+
+        # Check business rule: phone uniqueness
         is_phone_exists = await cls.app_svc_container.users_service.is_exists(
-            filter_data={"phone": validated_data.phone}
+            filter_data={"phone": phone_code_vo.phone}
         )
         if is_phone_exists:
             raise AlreadyExistsError(
                 message="Already exists",
                 details=[
-                    {"key": "phone", "value": mask_string(validated_data.phone, keep_start=2, keep_end=2)},
+                    {"key": "phone", "value": mask_string(phone_code_vo.phone, keep_start=2, keep_end=2)},
                 ],
             )
-        user_dto = await cls.create(validated_data.to_dict(), is_return_require=True, out_dataclass=UserShortDTO)
+
+        # Prepare persistence data
+        data = {
+            "phone": phone_code_vo.phone,
+            "verification_code": phone_code_vo.verification_code,
+        }
+        user_dto = await cls.create(data, is_return_require=True, out_dataclass=UserShortDTO)
         return user_dto

src/app/domain/auth/value_objects/__init__.py

@@ -1,4 +1,4 @@
-from src.app.domain.auth.value_objects.jwt_vob import (
+from src.app.domain.auth.value_objects.jwt_vo import (
     TokenType,
     TokenPayload,
     TokenPair,

src/app/domain/auth/value_objects/jwt_vo.py

@@ -0,0 +1,65 @@
+from dataclasses import dataclass
+from datetime import datetime
+from enum import Enum
+from typing import Optional
+
+
+class TokenType(str, Enum):
+    ACCESS = "access"
+    REFRESH = "refresh"
+
+
+@dataclass(frozen=True)
+class TokenPayload:
+    """Value object representing JWT token payload data."""
+
+    uuid: str
+    sid: str
+    token_type: TokenType
+    exp: datetime
+
+    def to_dict(self) -> dict:
+        return {
+            "uuid": self.uuid,
+            "sid": self.sid,
+        }
+
+
+@dataclass(frozen=True)
+class TokenPair:
+    """Value object representing a pair of access and refresh tokens."""
+
+    access_token: str
+    refresh_token: str
+
+    def to_dict(self) -> dict:
+        return {
+            "access": self.access_token,
+            "refresh": self.refresh_token,
+        }
+
+
+@dataclass(frozen=True)
+class DecodedToken:
+    """Value object representing decoded token data."""
+
+    uuid: str
+    sid: str
+    token_type: TokenType
+    exp: Optional[datetime] = None
+
+    @classmethod
+    def from_payload(cls, payload: dict, token_type: TokenType) -> "DecodedToken":
+        user_data = payload.get("user", {})
+        return cls(
+            uuid=user_data.get("uuid", ""),
+            sid=user_data.get("sid", ""),
+            token_type=token_type,
+            exp=payload.get("exp"),
+        )
+
+    def to_dict(self) -> dict:
+        return {
+            "uuid": self.uuid,
+            "sid": self.sid,
+        }

src/app/domain/users/value_objects/users_vo.py

@@ -0,0 +1,86 @@
+import re
+from dataclasses import dataclass
+
+from src.app.domain.common.exceptions import ValidationError
+from src.app.domain.common.utils.common import mask_string
+
+
+@dataclass(frozen=True)
+class EmailPasswordPair:
+    """Value object representing a pair of email and password"""
+
+    email: str
+    password: str
+
+    def __post_init__(self) -> None:
+        self.__validate_email(value=self.email)
+        self.__validate_password(value=self.password)
+
+    @staticmethod
+    def __validate_email(value: str) -> None:
+        # TODO: implement validation
+        pass
+
+    @staticmethod
+    def __validate_password(value: str) -> None:
+        details = [
+            {"key": "password", "value": mask_string(value, keep_start=1, keep_end=1)},
+        ]
+        if len(value) < 8:
+            raise ValidationError(message="Must be at least 8 characters long", details=details)
+
+        # Check for at least one uppercase letter
+        if not re.search(r"[A-Z]", value):
+            raise ValidationError(message="Must contain at least one uppercase letter", details=details)
+
+        # Check for at least one lowercase letter
+        if not re.search(r"[a-z]", value):
+            raise ValidationError(message="Must contain at least one lowercase letter", details=details)
+
+        # Check for at least one digit
+        if not re.search(r"[0-9]", value):
+            raise ValidationError(message="Must contain at least one digit", details=details)
+
+        # Check for at least one special character
+        if not re.search(r"[!@#$%^&*(),.?\":{}|<>]", value):
+            raise ValidationError(message="Must contain at least one special character", details=details)
+
+    def to_dict(self) -> dict:
+        return {
+            "email": self.email,
+            "password_hashed": self.password,
+        }
+
+
+@dataclass(frozen=True)
+class PhoneNumberCodePair:
+    """Value object representing a pair of phone number and code"""
+
+    phone: str
+    verification_code: str
+
+    def __post_init__(self) -> None:
+        self.__validate_phone(value=self.phone)
+        self.__validate_verification_code(value=self.verification_code)
+
+    @staticmethod
+    def __validate_phone(value: str) -> None:
+        pattern = r"^\+?\d{1,3}[-.\s]?\(?\d{1,4}\)?[-.\s]?\d{1,4}[-.\s]?\d{1,9}$"
+        match = re.match(pattern, value)
+        if not match or 8 > len(value) or len(value) > 16:
+            raise ValidationError(
+                message="Invalid value",
+                details=[{"key": "phone", "value": mask_string(value, keep_start=2, keep_end=2)}],
+            )
+
+    pass
+
+    @staticmethod
+    def __validate_verification_code(value: str) -> None:
+        pass
+
+    def to_dict(self) -> dict:
+        return {
+            "phone": self.phone,
+            "verification_code": self.verification_code,
+        }

src/app/interfaces/api/v1/endpoints/auth/resources.py

@@ -16,12 +16,12 @@
 @router.post(path="/sign-up/", response_model=SignupResp, name="sign-up")
 async def sign_up(data: Annotated[SignUpReq, Body()]) -> dict:
 
-    user = await app_svc_container.users_service.create_user_by_email(
+    user_dto = await app_svc_container.users_service.create_user_by_email(
         email=data.email,
         password=data.password,
     )
-    assert user is not None
-    return asdict(user)
+    assert user_dto is not None
+    return asdict(user_dto)
 
 
 @router.post(path="/tokens/", response_model=TokenResp, name="tokens pair")
@@ -30,13 +30,13 @@ async def tokens(
 ) -> dict:
     """Get new access, refresh tokens [Based on email, password]"""
 
-    user = await app_svc_container.auth_service.get_auth_user_by_email_password(
+    user_dto = await app_svc_container.auth_service.get_auth_user_by_email_password(
         email=data.email, password=data.password
     )
 
-    token_pair = app_svc_container.auth_service.create_tokens_for_user(uuid=str(user.uuid))
+    token_pair = app_svc_container.auth_service.create_tokens_for_user(uuid=str(user_dto.uuid))
     tokens_data = {
-        "user_data": {"uuid": str(user.uuid)},
+        "user_data": {"uuid": str(user_dto.uuid)},
         "access": token_pair.access_token,
         "refresh": token_pair.refresh_token,
     }
@@ -48,9 +48,9 @@ async def tokens(
 async def tokens_refreshed(auth_api_key: str = Depends(validate_api_key)) -> dict:
     """Get new access, refresh tokens [Granted by refresh token in header]"""
 
-    user, token_pair = await app_svc_container.auth_service.refresh_tokens(auth_api_key)
+    user_dto, token_pair = await app_svc_container.auth_service.refresh_tokens(auth_api_key)
     tokens_data = {
-        "user_data": {"uuid": str(getattr(user, "uuid", ""))},
+        "user_data": {"uuid": str(getattr(user_dto, "uuid", ""))},
         "access": token_pair.access_token,
         "refresh": token_pair.refresh_token,
     }