FitTrackee/fittrackee/users/utils/admin.py

import secrets
from typing import Optional, Tuple

from sqlalchemy import func

from fittrackee import db

from ..exceptions import (
    InvalidEmailException,
    UserControlsException,
    UserCreationException,
    UserNotFoundException,
)
from ..models import User
from ..utils.controls import is_valid_email, register_controls


class UserManagerService:
    def __init__(self, username: str):
        self.username = username

    def _get_user(self) -> User:
        user = User.query.filter_by(username=self.username).first()
        if not user:
            raise UserNotFoundException()
        return user

    def _update_admin_rights(self, user: User, is_admin: bool) -> None:
        user.admin = is_admin
        if is_admin:
            self._activate_user(user)

    @staticmethod
    def _activate_user(user: User) -> None:
        user.is_active = True
        user.confirmation_token = None

    @staticmethod
    def _reset_user_password(user: User) -> str:
        new_password = secrets.token_urlsafe(30)
        user.password = user.generate_password_hash(new_password)
        return new_password

    @staticmethod
    def _update_user_email(
        user: User, new_email: str, with_confirmation: bool
    ) -> None:
        if not is_valid_email(new_email):
            raise InvalidEmailException('valid email must be provided')
        if user.email == new_email:
            raise InvalidEmailException(
                'new email must be different than current email'
            )
        if with_confirmation:
            user.email_to_confirm = new_email
            user.confirmation_token = secrets.token_urlsafe(30)
        else:
            user.email = new_email

    def update(
        self,
        is_admin: Optional[bool] = None,
        activate: bool = False,
        reset_password: bool = False,
        new_email: Optional[str] = None,
        with_confirmation: bool = True,
    ) -> Tuple[User, bool, Optional[str]]:
        user_updated = False
        new_password = None
        user = self._get_user()

        if is_admin is not None:
            self._update_admin_rights(user, is_admin)
            user_updated = True

        if activate:
            self._activate_user(user)
            user_updated = True

        if reset_password:
            new_password = self._reset_user_password(user)
            user_updated = True

        if new_email is not None:
            self._update_user_email(user, new_email, with_confirmation)
            user_updated = True

        db.session.commit()
        return user, user_updated, new_password

    def create_user(
        self,
        email: str,
        password: Optional[str] = None,
        check_email: bool = False,
    ) -> Tuple[Optional[User], Optional[str]]:
        if not password:
            password = secrets.token_urlsafe(30)

        ret = register_controls(self.username, email, password)

        if ret != '':
            raise UserControlsException(ret)

        user = User.query.filter(
            func.lower(User.username) == func.lower(self.username)
        ).first()
        if user:
            raise UserCreationException(
                'sorry, that username is already taken'
            )

        # if a user exists with same email address, no error is returned
        # since a user has to confirm his email to activate his account
        user = User.query.filter(
            func.lower(User.email) == func.lower(email)
        ).first()
        if user:
            if check_email:
                raise UserCreationException(
                    'This user already exists. No action done.'
                )
            return None, None

        new_user = User(username=self.username, email=email, password=password)
        new_user.timezone = 'Europe/Paris'
        new_user.date_format = 'MM/dd/yyyy'
        new_user.confirmation_token = secrets.token_urlsafe(30)
        db.session.add(new_user)
        db.session.flush()

        return new_user, password

    def create(
        self,
        email: str,
        password: Optional[str] = None,
    ) -> Tuple[Optional[User], Optional[str]]:
        try:
            new_user, password = self.create_user(
                email, password, check_email=True
            )
            if new_user:
                new_user.language = 'en'
                db.session.commit()
        except UserControlsException as e:
            raise UserCreationException(str(e))
        return new_user, password