cx-notes/assets/users_manager.py

import sqlmodel
import sqlalchemy

from .user import user
from .user_validator import user_validator
from .user_validator import user_validation_result
from .user_validator import user_validation_exception

class users_manager:
    def __init__(self, engine: sqlalchemy.engine.base.Engine):
        self.__engine = engine

    @property
    def session(self) -> sqlmodel.Session:
        return sqlmodel.Session(self.__engine)
    
    def register(self, target: user) -> user:
        validation_result = self.__validate_user(target)

        if validation_result != user_validation_result.VALID:
            raise user_validation_exception(validation_result)

        if self.nick_taken(target.nick):
            raise user_validation_exception(
                user_validation_result.NICK_ALREADY_TAKEN
            )

        if validator.has_id:
            raise Exception("Can not register user which already exists.")

        with self.session as session:
            session.add(target)
            session.commit()
            session.refresh(target)

        return target

    def load(self, target: int) -> user | None:
        query = sqlmodel
        query = query.select(user)
        query = query.where(user.id == target)

        result = None

        with self.session as session:
            result = session.exec(query).first()

        return result

    def who_take_nick(self, target: str) -> int | None:
        query = sqlmodel
        query = query.select(user)
        query = query.where(user.nick == target)

        with self.session as session:
            result = session.exec(query).first()

            if result is None:
                return None

            return result.id

    def nick_taken(self, target: str) -> bool:
        return self.who_take_nick(target) != None

    def __validate_user(self, target: user) -> user_validation_result:
        validator = user_validator(target)

        if not validator.has_valid_nick:
            return validator.nick_validation_result

        if not validator.has_valid_secret:
            return validator.secret_validation_result

        return user_validation_result.VALID

    def save(self, target: user) -> user:
        validation_result = self.__validate_user(target)

        if validation_result != user_validation_result.VALID:
            raise user_validation_exception(validation_result)
        
        if self.who_take_nick(target.nick) != target.id:
            raise user_validation_exception(
                user_validation_result.NICK_ALREADY_TAKEN
            )

        if not target.has_id:
            raise Exception("Register user before save it.")

        with self.session as session:
            session.add(target)
            session.commit()
            session.refresh(target)

    def login(self, nick: str, secret: str) -> user | None:
        nick_validation = user_validator.validate_nick(nick)

        if nick_validation != user_validation_result.VALID:
            raise user_validation_exception(nick_validation)

        secret_validation = user_validator.validate_secret(secret)
        
        if secret_validation != user_validation_result.VALID:
            raise user_validation_exception(secret_validation)
        
        result = None

        with self.session as session:
            query = sqlmodel
            query = query.select(user)
            query = query.where(user.nick == nick)
            query = query.where(user.secret == secret)

            result = session.exec(query).first()

        return result

    def remove(self, target: user) -> None:
        with self.session as session:
            session.delete(target)
            session.commit()