backend/src/api/services/auth.py - master - FlysAt/DataFoundry

backend/src/api/services/auth.py
96 строк3 KB
AlexanderShmygol
refactoring
07 апр 2026, 13:55
07 апр 2026, 13:555cff70c
from fastapi import HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from src.api.repositories.user import UserRepository
from src.core.security import (
    create_access_token,
    create_refresh_token,
    decode_token,
    hash_password,
    verify_password,
)
from src.models.user import User
from src.schemas.auth import RefreshTokenRequestSchema, TokenPairSchema, UserLoginSchema, UserRegisterSchema


class AuthService:
    ALLOWED_ROLES = {"executor", "customer", "user"}

    def __init__(self, db: AsyncSession) -> None:
        self.users = UserRepository(db)

    async def register_user(self, payload: UserRegisterSchema) -> User:
        invalid_roles = set(payload.roles) - self.ALLOWED_ROLES
        if invalid_roles:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"Invalid roles: {', '.join(invalid_roles)}",
            )

        existing_user = await self.users.get_by_email(payload.email)
        if existing_user:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="User already exists",
            )

        user = User(
            email=payload.email,
            hashed_password=hash_password(payload.password),
            full_name=payload.full_name,
            roles=payload.roles or ["user"],
        )
        return await self.users.create(user)

    async def login_user(self, payload: UserLoginSchema) -> TokenPairSchema:
        user = await self.users.get_by_email(payload.email)
        if not user or not verify_password(payload.password, user.hashed_password):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid email or password",
            )

        return self._build_token_pair(user)

    async def refresh_tokens(self, payload: RefreshTokenRequestSchema) -> TokenPairSchema:
        try:
            token_data = decode_token(payload.refresh_token)
        except ValueError as exc:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid or expired refresh token",
            ) from exc

        if token_data.get("type") != "refresh":
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid token type",
            )

        user_id = token_data.get("sub")
        if user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid token payload",
            )

        user = await self.users.get_by_id(int(user_id))
        if not user:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="User not found",
            )

        return self._build_token_pair(user)

    @staticmethod
    def _build_token_pair(user: User) -> TokenPairSchema:
        token_payload = {
            "sub": str(user.id),
            "email": user.email,
            "roles": user.roles,
        }
        return TokenPairSchema(
            access_token=create_access_token(token_payload),
            refresh_token=create_refresh_token({"sub": str(user.id)}),
        )