Source code for py_abac.storage.base

"""
    Policy Storage abstract class
"""

from abc import ABCMeta, abstractmethod
from typing import Generator, Union

from ..policy import Policy


class Storage(metaclass=ABCMeta):
    """
        Base class for policy storage
    """

    @abstractmethod
    def add(self, policy: Policy):
        """
            Store a policy
        """
        raise NotImplementedError()

    @abstractmethod
    def get(self, uid: str) -> Union[Policy, None]:
        """
            Get specific policy
        """
        raise NotImplementedError()

    @abstractmethod
    def get_all(self, limit: int, offset: int) -> Generator[Policy, None, None]:
        """
            Retrieve all the policies within a window
        """
        raise NotImplementedError()

    @abstractmethod
    def get_for_target(
            self,
            subject_id: str,
            resource_id: str,
            action_id: str
    ) -> Generator[Policy, None, None]:
        """
            Get all policies for given target IDs.
        """
        # TODO: Add policy retrieval caching
        raise NotImplementedError()

    @abstractmethod
    def update(self, policy: Policy):
        """
            Update a policy
        """
        raise NotImplementedError()

    @abstractmethod
    def delete(self, uid: str):
        """
            Delete a policy
        """
        raise NotImplementedError()

    @staticmethod
    def _check_limit_and_offset(limit: int, offset: int):
        if limit < 0:
            raise ValueError("Limit can't be negative")
        if offset < 0:
            raise ValueError("Offset can't be negative")