Atheneum/server/atheneum/service/user_service.py

"""Service to handle user operations."""
import logging
from datetime import datetime
from typing import Optional, Dict, Callable, Any

from atheneum.db import db
from atheneum.model import User
from atheneum.utility import authentication_utility
from atheneum.service.serialization_service import (
    BaseSerializer,
    register_serializer
)

LOGGER = logging.getLogger(__name__)


class UserSerializer(BaseSerializer):
    """Serialize User model."""

    def _serializers(self) -> Dict[str, Callable[[], Any]]:
        """Define the fields and the accompanying serializer factory."""
        return {
            'name': self.serialize_name,
            'creationTime': self.serialize_creation_time,
            'lastLoginTime': self.serialize_last_login_time,
            'version': self.serialize_version,
            'role': self.serialize_role,
        }

    def serialize_name(self) -> str:
        """User name."""
        return self.model.name

    def serialize_creation_time(self) -> datetime:
        """User creation time."""
        return self.model.creation_time

    def serialize_last_login_time(self) -> datetime:
        """User last login time."""
        return self.model.last_login_time

    def serialize_version(self) -> int:
        """User version."""
        return self.model.version

    def serialize_role(self) -> str:
        """User role."""
        return self.model.role


register_serializer(User.__name__, UserSerializer)


def register(name: str, password: str, role: str) -> User:
    """
    Register a new user.

    :param name: Desired user name. Must be unique and not already registered
    :param password: Password to be hashed and stored for the user
    :param role: Role to assign the user [ROLE_USER, ROLE_ADMIN]
    :return:
    """
    pw_hash, pw_revision = authentication_utility.get_password_hash(password)

    new_user = User(
        name=name,
        role=role,
        password_hash=pw_hash,
        password_revision=pw_revision,
        creation_time=datetime.now(),
        version=0)
    db.session.add(new_user)
    db.session.commit()

    LOGGER.info('Registered new user: %s with role: %s', name, role)
    return new_user


def delete(user: User) -> bool:
    """
    Delete a user.

    :param user:
    :return:
    """
    existing_user = db.session.delete(user)
    if existing_user is None:
        db.session.commit()
        return True
    return False


def update_last_login_time(user: User) -> None:
    """
    Bump the last login time for the user.

    :param user:
    :return:
    """
    if user is not None:
        user.last_login_time = datetime.now()
        db.session.commit()


def update_password(user: User, password: str) -> None:
    """
    Change the user password.

    :param user:
    :param password:
    :return:
    """
    pw_hash, pw_revision = authentication_utility.get_password_hash(
        password)
    user.password_hash = pw_hash
    user.password_revision = pw_revision
    db.session.commit()


def find_by_name(name: str) -> Optional[User]:
    """
    Find a user by name.

    :param name:
    :return:
    """
    return User.query.filter_by(name=name).first()