cx-keys/assets/user_loader.py

import sqlmodel

from .user import user
from .user import user_builder
from .password import password

class user_loader(sqlmodel.Session):
    def login(self, nick: str, secret: str) -> user | None:
        """
        This get new user by nick and password. If user does not exists,
        then return None. It also return None when user password is not
        correct.

        nick: str - Nick of the user
        secret: str - Password of the user
        return: user | None - Logged user, or None when not exists
        """

        result = self.get_by_nick(nick)

        if result is None:
            return None

        if not password(secret).validate(result.password):
            return None

        return result

    def register(self, target: user) -> bool:
        """
        This function register new user. It also check that user with given
        apikey or nick not exists. If user had been created, then return True.
        When something went wront, return False.

        target: user - User to register in database
        return: bool - True when registered new user, False when failed
        """

        if target.id is not None:
            return False

        if target.apikey is None or target.nick is None:
            return False

        if target.password is None:
            return False

        if self.nick_in_use(target.nick):
            return False

        if self.apikey_in_use(target.apikey):
            return False

        self.add(target)
        self.commit()
        self.refresh(target)

        return True

    def unregister(self, target: user) -> bool:
        """
        This function remove user from database.

        target: user - User to remove from database
        return: bool - True when removed, Falsd when something failed
        """

        if not self.is_registered(target):
            return False

        self.delete(target)
        self.commit()

        return True

    def save(self, target: user) -> bool:
        """
        This function save already registered user to database, after changes.

        target: user - User to save
        return: bool - True when saved successfull, False when failed
        """

        if not self.is_registered(user):
            return False

        self.add(target)
        self.commit()
        self.refresh(target)

        return True

    def nick_in_use(self, nick: str) -> bool:
        """
        This function check that nick is already in use.

        nick: str - Nick to check
        return: bool - True when nick in use, False if not
        """

        return self.get_by_nick(nick) is not None

    def get_by_apikey(self, apikey: str) -> user | None:
        """
        This get user from database by apikey, or return None when not exists.

        apikey: str - Apikey of the user to load
        return: user | None - User with given apikey, or none when not exists
        """

        query = sqlmodel.select(user).where(user.apikey == apikey).limit(1)
        result = self.exec(query)

        return result.first()

    def apikey_in_use(self, apikey: str) -> bool:
        """
        This function check that apikey is already in use.

        apikey: str - Apikey to check
        return: bool - True when apikey is in use or False when not
        """

        return self.get_by_apikey(apikey) is not None

    def get_by_nick(self, nick: str) -> user | None:
        """
        This get user from database by nick.

        nick: str - Nick of the user to load
        return: user | None - Loaded user or None when not exists.
        """

        query = sqlmodel.select(user).where(user.nick == nick).limit(1)
        result = self.exec(query)

        return result.first()

    def get_by_id(self, id: int) -> user | None:
        """
        This function select user from database by ID.

        id: int - ID of the user to select
        return: user | None - Selected user or None when not exists
        """

        query = sqlmodel.select(user).where(user.id == id).limit(1)
        result = self.exec(query)

        return result.first()

    def is_registered(self, target: user) -> bool:
        """
        This function check that user is registered.

        target: user - User to check
        return: bool - True when user is registered, False when not
        """

        if target.id is None:
            return False

        return self.get_by_id(target.id) is not None