cx-keys/assets/secret_loader.py

import typing
import sqlmodel
import sqlalchemy.engine.base

from .user import user
from .secret import secret

class secret_loader(sqlmodel.Session):
    """
    This class is responsible for loading and managing secrets in the
    database.
    """

    def __init__(
        self, 
        connection: sqlalchemy.engine.base.Engine, 
        owner: user
    ) -> None:
        """
        This create new loader. It require connection to the database, and 
        user, which own secrets to work on.
        
        Parameters:
            connection (sqlalchemy.engine.base.Engine): Connection to database
            owner (user): User which own secrets to work on
        """

        super().__init__(connection)

        if not owner.in_database:
            raise Exception("User to build loaded for not exists.")

        self.__owner = owner

    @property
    def owner(self) -> user:
        """ This return owner of secrets which work on. """

        return self.__owner

    def append(self, target: secret) -> bool:
        """
        This append new secret to database. Secret must have same owner, as 
        set in loader. When owner is not set, user from loader would be used.
        When user is not correct, that mean is other in secret and in loader, 
        TypeError would be raised. When secret name is already in use, then
        it return False. When item is already in database, Exception would be
        raised. When secret is not ready, that mean 

        Parameters:
            target (secret): New secret to insert in database

        Returns:
            (bool): True when append successfull, False when failed
        """
        
        if target.owner is None:
            target.owner = self.owner    
        elif target.owner != self.owner.id:
            raise TypeError("Owner of secret is other than in loader.")
        
        if target.in_database: 
            raise Exception("Secret is already in database. Use update.")

        if not target.ready:
            raise Exception("Secret is not ready to append.")

        if self.load_by_name(target.name) is not None:
            return False

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

        return True

    @property
    def __select(self) -> sqlmodel.sql.expression.Select:
        """ It create new selecr query with set owner statement. """

        return sqlmodel.select(secret).where(secret.owner == self.owner.id)

    def search_for_domain(self, target: str) -> typing.Iterable[secret]:
        """ 
        This search in secret database for secrets by domain. It return 
        iterator which items is in secret type.

        Parameters:
            target (str): Domain name to search

        Returns:
            (Iterable[secret]): Iterator with found secrets
        """

        target = "%" + target + "%"
        query = self.__select.where(secret.domain.like(target))
        result = self.exec(query)

        for item in result:
            yield item

    def search_for_name(self, target: str) -> typing.Iterable[secret]:
        """
        This search in secrets database for secrets which have names like 
        given. It returns iterator with found result.

        Parameters:
            target (str): Name to search in database
        
        Returns:
            (Iterable[secret]): Found secrets
        """

        target = "%" + target.upper() + "%"
        query = self.__select.where(secret.name.like(target))
        result = self.exec(query)

        for item in result:
            yield item

    def name_in_use(self, target: str) -> bool:
        """
        This check that name is already in use.

        Parameters:
            target (str): Name to check

        Returns:
            (bool): True when name is user, False when not
        """

        return self.load_by_name(target) is not None

    def load_by_name(self, target: str) -> secret | None:
        """
        It load single secret by name. When secret not exists, return None.

        Parameters:
            target (str): Name of the secret

        Returns:
            (secret | None): Loaded secret or None when not exists
        """

        target = target.upper()
        query = self.__select.where(secret.name == target)
        result = self.exec(query)

        return result.first()

    def drop(self, target: secret) -> bool:
        """
        This remove secret from database. When secret not exists in database
        return False. Whem all went good, return True.

        Parameters:
            target (secret): Secret to drop

        Returns:
            (bool): True when remove successfull, False when not
        """

        if not target.in_database:
            return False

        self.delete(target)
        self.commit()

        return True

    def clear(self) -> None:
        """
        This drop all secrets of the user. It is useable before user remove.
        """

        query = self.__select
        result = self.exec(query)

        for item in result:
            self.delete(item)

        self.commit()

    def update(self, target: secret) -> bool:
        """
        This update secret in the database. When secret not exists in 
        database. When owner is not set propertly, raise TypeError. When
        target is not in database, or secret is not ready, raise Exception.
        When name is already in use by other secret, it return False.

        Parameters:
            target (secret): Secret to update

        Returns:
            (bool): True when updated, False when name is in use
        """

        if target.owner is None:
            target.owner = self.owner
        elif target.owner != self.owner.id:
            raise TypeError("Owner of the secret is other than in loader.")

        if not target.in_database:
            raise Exception("Target is not in database. Use append.")

        if not target.ready:
            return Exception("Target is not ready to update.")

        check = self.load_by_name(target.name)

        if check is not None and check.id != target.id:
            return False

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

        return True