Module lib.repository

Repository types reside in this module

"""
Repository types reside in this module
"""
from abc import abstractmethod

layers = []


def _call_layers(action, repo_cls, model_id=None, new_model=None, id_attr='id'):  # pylint: disable=too-many-branches
    """
    Utility method to call all layers

    Pylint: too-many-branches disabled because a dictionary dispatch system would not work here.
    :param action: str. Identifies what layer action to take
    :param repo_cls: The repository class where the action is done.
    :param model_id: Optional. ID desired
    :param new_model: Optional. If a new model is made or read, this is handed to the layer method
    :param id_attr: Optional. Name of ID attribute (default is 'id')
    :return: None
    """
    if model_id is not None and new_model is not None:
        raise ValueError('Either model_id or new_model needs to be specified!')

    if action == 'create':
        for layer in layers:
            layer.on_create(repo_cls, new_model)
        return
    if action == 'read_one':
        for layer in layers:
            layer.on_read_one(repo_cls, new_model)
        return
    if action == 'read_many':
        for model in new_model:
            for layer in layers:
                layer.on_read_one(repo_cls, model)
        return
    if action == 'update':
        for layer in layers:
            layer.on_update(repo_cls, new_model, id_attr=id_attr)
        return
    if action == 'destroy':
        for layer in layers:
            layer.on_destroy(repo_cls, model_id, id_attr=id_attr)
    else:
        raise ValueError('Invalid action specified')


class CanMutateData:
    """
    Attribute for Repositories that allow for data mutation after reading
    """

    @classmethod
    @abstractmethod
    def after_read(cls, model_read):
        """
        Called after a model is read
        :param model_read: the model object read from data source
        :return: raises Exception if interrupt is desired
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def after_read_many(cls, models_read):
        """
        Called after many models are read
        :param models_read: model objects read from data source
        :return: raises Exception if interrupt is desired
        """
        raise NotImplementedError


class Repository(CanMutateData):
    """
    Base Repository Class for basic CRUD operations
    """
    id_attr: str = 'id'

    @classmethod
    def on_create(cls, model):
        """
        Calls the on_create method on all registered layers

        Repository implementations MUST call this manually, as layers
        are an opt-in process
        :param model: model newly created
        :return: None
        """
        _call_layers('create', cls, new_model=model)

    @classmethod
    def on_update(cls, model):
        """
        Calls the on_update method on all registered layers

        Repository implementations MUST call this manually, as layers
        are an opt-in process
        :param model: model newly updated
        :return: None
        """
        _call_layers('update', cls, new_model=model, id_attr=cls.id_attr)

    @classmethod
    def on_destroy(cls, model_id):
        """
        Calls the on_destroy method on all registered layers

        Repository implementations MUST call this manually, as layers
        are an opt-in process
        :param model_id: ID of deleted model (or to-be deleted)
        :return: None
        """
        _call_layers('destroy', cls, model_id=model_id, id_attr=cls.id_attr)

    @classmethod
    @abstractmethod
    def create(cls, model):
        """
        Creates the model in the data sink of choice
        :param model: instance with data
        :return: model
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def read(cls, model_id):
        """
        Read a single Model by it's ID
        :param model_id: model ID
        :param id_attr: Optional. Name of ID attribute (default is 'id')
        :return: model
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def read_by(cls, filters=None):
        """
        Reads all models that fit into the filters
        :return: list of models
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def read_all(cls):
        """
        Reads all models
        :return: list of models
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def update(cls, model):
        """
        Uses the 'id' column to update the model
        :param model: model instance
        :return: model
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def destroy(cls, model_id):
        """
        Deletes the model from the data sink
        :param model_id: model's ID to destroy
        :return: None
        """
        raise NotImplementedError

    @classmethod
    def after_read(cls, model_read):
        _call_layers('read_one', cls, new_model=model_read, id_attr=cls.id_attr)

    @classmethod
    def after_read_many(cls, models_read):
        _call_layers('read_many', cls, new_model=models_read)

    @property
    @classmethod
    @abstractmethod
    def resource_uri(cls):
        """
        This is a helper to make the table/file/URI name more obvious.

        For example, the table() function in a DatabaseRepo should use
        this property to find the table name
        :return: Resource URI
        """
        raise AttributeError('resource_uri: No resource URI set!')

Sub-modules

lib.repository.db

Database Repository with helper methods for setup

lib.repository.validator

Contains all the validation logic for the backend

Classes

class CanMutateData

Attribute for Repositories that allow for data mutation after reading

Expand source code

class CanMutateData:
    """
    Attribute for Repositories that allow for data mutation after reading
    """

    @classmethod
    @abstractmethod
    def after_read(cls, model_read):
        """
        Called after a model is read
        :param model_read: the model object read from data source
        :return: raises Exception if interrupt is desired
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def after_read_many(cls, models_read):
        """
        Called after many models are read
        :param models_read: model objects read from data source
        :return: raises Exception if interrupt is desired
        """
        raise NotImplementedError

Subclasses

• Repository

Static methods

def after_read(model_read)

Called after a model is read :param model_read: the model object read from data source :return: raises Exception if interrupt is desired

Expand source code

@classmethod
@abstractmethod
def after_read(cls, model_read):
    """
    Called after a model is read
    :param model_read: the model object read from data source
    :return: raises Exception if interrupt is desired
    """
    raise NotImplementedError

def after_read_many(models_read)

Called after many models are read :param models_read: model objects read from data source :return: raises Exception if interrupt is desired

Expand source code

@classmethod
@abstractmethod
def after_read_many(cls, models_read):
    """
    Called after many models are read
    :param models_read: model objects read from data source
    :return: raises Exception if interrupt is desired
    """
    raise NotImplementedError

class Repository

Base Repository Class for basic CRUD operations

Expand source code

class Repository(CanMutateData):
    """
    Base Repository Class for basic CRUD operations
    """
    id_attr: str = 'id'

    @classmethod
    def on_create(cls, model):
        """
        Calls the on_create method on all registered layers

        Repository implementations MUST call this manually, as layers
        are an opt-in process
        :param model: model newly created
        :return: None
        """
        _call_layers('create', cls, new_model=model)

    @classmethod
    def on_update(cls, model):
        """
        Calls the on_update method on all registered layers

        Repository implementations MUST call this manually, as layers
        are an opt-in process
        :param model: model newly updated
        :return: None
        """
        _call_layers('update', cls, new_model=model, id_attr=cls.id_attr)

    @classmethod
    def on_destroy(cls, model_id):
        """
        Calls the on_destroy method on all registered layers

        Repository implementations MUST call this manually, as layers
        are an opt-in process
        :param model_id: ID of deleted model (or to-be deleted)
        :return: None
        """
        _call_layers('destroy', cls, model_id=model_id, id_attr=cls.id_attr)

    @classmethod
    @abstractmethod
    def create(cls, model):
        """
        Creates the model in the data sink of choice
        :param model: instance with data
        :return: model
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def read(cls, model_id):
        """
        Read a single Model by it's ID
        :param model_id: model ID
        :param id_attr: Optional. Name of ID attribute (default is 'id')
        :return: model
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def read_by(cls, filters=None):
        """
        Reads all models that fit into the filters
        :return: list of models
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def read_all(cls):
        """
        Reads all models
        :return: list of models
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def update(cls, model):
        """
        Uses the 'id' column to update the model
        :param model: model instance
        :return: model
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def destroy(cls, model_id):
        """
        Deletes the model from the data sink
        :param model_id: model's ID to destroy
        :return: None
        """
        raise NotImplementedError

    @classmethod
    def after_read(cls, model_read):
        _call_layers('read_one', cls, new_model=model_read, id_attr=cls.id_attr)

    @classmethod
    def after_read_many(cls, models_read):
        _call_layers('read_many', cls, new_model=models_read)

    @property
    @classmethod
    @abstractmethod
    def resource_uri(cls):
        """
        This is a helper to make the table/file/URI name more obvious.

        For example, the table() function in a DatabaseRepo should use
        this property to find the table name
        :return: Resource URI
        """
        raise AttributeError('resource_uri: No resource URI set!')

Ancestors

• CanMutateData

Subclasses

• DatabaseRepository

Class variables

var id_attr : str

Static methods

def create(model)

Creates the model in the data sink of choice :param model: instance with data :return: model

Expand source code

@classmethod
@abstractmethod
def create(cls, model):
    """
    Creates the model in the data sink of choice
    :param model: instance with data
    :return: model
    """
    raise NotImplementedError

def destroy(model_id)

Deletes the model from the data sink :param model_id: model's ID to destroy :return: None

Expand source code

@classmethod
@abstractmethod
def destroy(cls, model_id):
    """
    Deletes the model from the data sink
    :param model_id: model's ID to destroy
    :return: None
    """
    raise NotImplementedError

def on_create(model)

Calls the on_create method on all registered layers

Repository implementations MUST call this manually, as layers are an opt-in process :param model: model newly created :return: None

Expand source code

@classmethod
def on_create(cls, model):
    """
    Calls the on_create method on all registered layers

    Repository implementations MUST call this manually, as layers
    are an opt-in process
    :param model: model newly created
    :return: None
    """
    _call_layers('create', cls, new_model=model)

def on_destroy(model_id)

Calls the on_destroy method on all registered layers

Repository implementations MUST call this manually, as layers are an opt-in process :param model_id: ID of deleted model (or to-be deleted) :return: None

Expand source code

@classmethod
def on_destroy(cls, model_id):
    """
    Calls the on_destroy method on all registered layers

    Repository implementations MUST call this manually, as layers
    are an opt-in process
    :param model_id: ID of deleted model (or to-be deleted)
    :return: None
    """
    _call_layers('destroy', cls, model_id=model_id, id_attr=cls.id_attr)

def on_update(model)

Calls the on_update method on all registered layers

Repository implementations MUST call this manually, as layers are an opt-in process :param model: model newly updated :return: None

Expand source code

@classmethod
def on_update(cls, model):
    """
    Calls the on_update method on all registered layers

    Repository implementations MUST call this manually, as layers
    are an opt-in process
    :param model: model newly updated
    :return: None
    """
    _call_layers('update', cls, new_model=model, id_attr=cls.id_attr)

def read(model_id)

Read a single Model by it's ID :param model_id: model ID :param id_attr: Optional. Name of ID attribute (default is 'id') :return: model

Expand source code

@classmethod
@abstractmethod
def read(cls, model_id):
    """
    Read a single Model by it's ID
    :param model_id: model ID
    :param id_attr: Optional. Name of ID attribute (default is 'id')
    :return: model
    """
    raise NotImplementedError

def read_all()

Reads all models :return: list of models

Expand source code

@classmethod
@abstractmethod
def read_all(cls):
    """
    Reads all models
    :return: list of models
    """
    raise NotImplementedError

def read_by(filters=None)

Reads all models that fit into the filters :return: list of models

Expand source code

@classmethod
@abstractmethod
def read_by(cls, filters=None):
    """
    Reads all models that fit into the filters
    :return: list of models
    """
    raise NotImplementedError

def update(model)

Uses the 'id' column to update the model :param model: model instance :return: model

Expand source code

@classmethod
@abstractmethod
def update(cls, model):
    """
    Uses the 'id' column to update the model
    :param model: model instance
    :return: model
    """
    raise NotImplementedError

Instance variables

var resource_uri

classmethod(function) -> method

Convert a function to be a class method.

A class method receives the class as implicit first argument, just like an instance method receives the instance. To declare a class method, use this idiom:

class C: @classmethod def f(cls, arg1, arg2, …): …

It can be called either on the class (e.g. C.f()) or on an instance (e.g. C().f()). The instance is ignored except for its class. If a class method is called for a derived class, the derived class object is passed as the implied first argument.

Class methods are different than C++ or Java static methods. If you want those, see the staticmethod builtin.

Inherited members

• CanMutateData:
  • after_read
  • after_read_many