Module licenseware.repository.interface.interface

Expand source code
from abc import ABCMeta, abstractmethod
from typing import List, Union
from marshmallow import Schema


class RepositoryInterface(metaclass=ABCMeta):
    # db_url param should be provided on constructor
    db_url = None

    @classmethod
    def __subclasshook__(cls, subclass):
        # Ensure that all methods and attrs are provided

        return (
                hasattr(subclass, 'db_url') and isinstance(subclass.db_url, str) or isinstance(subclass.db_url, str)
                and
                hasattr(subclass, 'fetch') and callable(subclass.fetch)
                and
                hasattr(subclass, 'fetch_one') and callable(subclass.fetch_one)
                and
                hasattr(subclass, 'fetch_by_id') and callable(subclass.fetch_by_id)
                and
                hasattr(subclass, 'fetch_many') and callable(subclass.fetch_many)
                and
                hasattr(subclass, 'insert') and callable(subclass.insert)
                and
                hasattr(subclass, 'insert_one') and callable(subclass.insert_one)
                and
                hasattr(subclass, 'insert_with_id') and callable(subclass.insert_with_id)
                and
                hasattr(subclass, 'insert_many') and callable(subclass.insert_many)
                and
                hasattr(subclass, 'update') and callable(subclass.update)
                and
                hasattr(subclass, 'update_one') and callable(subclass.update_one)
                and
                hasattr(subclass, 'update_on_id') and callable(subclass.update_on_id)
                and
                hasattr(subclass, 'update_many') and callable(subclass.update_many)
                and
                hasattr(subclass, 'delete') and callable(subclass.delete)
                and
                hasattr(subclass, 'delete_one') and callable(subclass.delete_one)
                and
                hasattr(subclass, 'delete_by_id') and callable(subclass.delete_by_id)
                and
                hasattr(subclass, 'delete_many') and callable(subclass.delete_many)
                and
                hasattr(subclass, 'count') and callable(subclass.count)

                # Fail
                or NotImplemented
        )

    # Fetching data

    @abstractmethod
    def fetch(self, schema_name: str, id: str = None, first: bool = False, limit: int = None, skip: int = None,
              **filters) -> List[dict]:
        """             
        """
        raise NotImplementedError

    @abstractmethod
    def fetch_one(self, schema_name: str, **filters) -> dict:
        """ Get first item match """
        return self.fetch(schema_name, first=True, **filters)

    @abstractmethod
    def fetch_by_id(self, schema_name: str, id: str) -> dict:
        """ Get first item match by id """
        return self.fetch(schema_name, first=True, id=id)

    @abstractmethod
    def fetch_many(self, schema_name: str, **filters) -> List[dict]:
        """ Get all items that matched  """
        return self.fetch(schema_name, **filters)

        # Inserting new data

    @abstractmethod
    def insert(self, schema: Schema, data: Union[dict, List[dict]]) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def insert_one(self, schema: Schema, data: dict = None, **data_kwargs) -> any:
        """ Insert dict to db """
        return self.insert(schema, data or data_kwargs)

    @abstractmethod
    def insert_with_id(self, schema: Schema, id: Union[str, int], data: dict = None, **data_kwargs) -> any:
        """ Insert dict with a specific 'id' """
        raise NotImplementedError

    @abstractmethod
    def insert_many(self, schema: Schema, data: List[dict], validate_percentage: float = 1.0) -> any:
        """
            TODO `validate_percentage`: what percentage of the list of data provided should be validated (0.5 is 50%)
        """
        return self.insert(schema, data)

    # Updating existing data

    @abstractmethod
    def update(self, schema_name: str, filters: dict, data: Union[dict, List[dict]], first: bool = False) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def update_one(self, filters: dict, data: dict, append: bool = False, schema: Schema = None) -> dict:
        """ """
        raise NotImplementedError

    @abstractmethod
    def update_on_id(self, schema_name: str, id: Union[str, int], data: dict) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def update_many(self, schema_name: str, filters: dict, data: List[dict], validate_percentage: float = 1.0) -> any:
        """
            `validate_percentage`: what percentage of the list of data provided should be validated (0.5 is 50% of items in list)
        """
        raise NotImplementedError

    # Deleting existing data

    @abstractmethod
    def delete(self, schema_name: str, filters: dict, first: bool = False) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def delete_one(self, schema_name: str, **filters) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def delete_by_id(elf, schema_name: str, id: str) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def delete_many(self, schema_name: str, **filters) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def count(self, schema_name: str, **filters) -> any:
        """ """
        raise NotImplementedError

Classes

class RepositoryInterface
Expand source code
class RepositoryInterface(metaclass=ABCMeta):
    # db_url param should be provided on constructor
    db_url = None

    @classmethod
    def __subclasshook__(cls, subclass):
        # Ensure that all methods and attrs are provided

        return (
                hasattr(subclass, 'db_url') and isinstance(subclass.db_url, str) or isinstance(subclass.db_url, str)
                and
                hasattr(subclass, 'fetch') and callable(subclass.fetch)
                and
                hasattr(subclass, 'fetch_one') and callable(subclass.fetch_one)
                and
                hasattr(subclass, 'fetch_by_id') and callable(subclass.fetch_by_id)
                and
                hasattr(subclass, 'fetch_many') and callable(subclass.fetch_many)
                and
                hasattr(subclass, 'insert') and callable(subclass.insert)
                and
                hasattr(subclass, 'insert_one') and callable(subclass.insert_one)
                and
                hasattr(subclass, 'insert_with_id') and callable(subclass.insert_with_id)
                and
                hasattr(subclass, 'insert_many') and callable(subclass.insert_many)
                and
                hasattr(subclass, 'update') and callable(subclass.update)
                and
                hasattr(subclass, 'update_one') and callable(subclass.update_one)
                and
                hasattr(subclass, 'update_on_id') and callable(subclass.update_on_id)
                and
                hasattr(subclass, 'update_many') and callable(subclass.update_many)
                and
                hasattr(subclass, 'delete') and callable(subclass.delete)
                and
                hasattr(subclass, 'delete_one') and callable(subclass.delete_one)
                and
                hasattr(subclass, 'delete_by_id') and callable(subclass.delete_by_id)
                and
                hasattr(subclass, 'delete_many') and callable(subclass.delete_many)
                and
                hasattr(subclass, 'count') and callable(subclass.count)

                # Fail
                or NotImplemented
        )

    # Fetching data

    @abstractmethod
    def fetch(self, schema_name: str, id: str = None, first: bool = False, limit: int = None, skip: int = None,
              **filters) -> List[dict]:
        """             
        """
        raise NotImplementedError

    @abstractmethod
    def fetch_one(self, schema_name: str, **filters) -> dict:
        """ Get first item match """
        return self.fetch(schema_name, first=True, **filters)

    @abstractmethod
    def fetch_by_id(self, schema_name: str, id: str) -> dict:
        """ Get first item match by id """
        return self.fetch(schema_name, first=True, id=id)

    @abstractmethod
    def fetch_many(self, schema_name: str, **filters) -> List[dict]:
        """ Get all items that matched  """
        return self.fetch(schema_name, **filters)

        # Inserting new data

    @abstractmethod
    def insert(self, schema: Schema, data: Union[dict, List[dict]]) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def insert_one(self, schema: Schema, data: dict = None, **data_kwargs) -> any:
        """ Insert dict to db """
        return self.insert(schema, data or data_kwargs)

    @abstractmethod
    def insert_with_id(self, schema: Schema, id: Union[str, int], data: dict = None, **data_kwargs) -> any:
        """ Insert dict with a specific 'id' """
        raise NotImplementedError

    @abstractmethod
    def insert_many(self, schema: Schema, data: List[dict], validate_percentage: float = 1.0) -> any:
        """
            TODO `validate_percentage`: what percentage of the list of data provided should be validated (0.5 is 50%)
        """
        return self.insert(schema, data)

    # Updating existing data

    @abstractmethod
    def update(self, schema_name: str, filters: dict, data: Union[dict, List[dict]], first: bool = False) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def update_one(self, filters: dict, data: dict, append: bool = False, schema: Schema = None) -> dict:
        """ """
        raise NotImplementedError

    @abstractmethod
    def update_on_id(self, schema_name: str, id: Union[str, int], data: dict) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def update_many(self, schema_name: str, filters: dict, data: List[dict], validate_percentage: float = 1.0) -> any:
        """
            `validate_percentage`: what percentage of the list of data provided should be validated (0.5 is 50% of items in list)
        """
        raise NotImplementedError

    # Deleting existing data

    @abstractmethod
    def delete(self, schema_name: str, filters: dict, first: bool = False) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def delete_one(self, schema_name: str, **filters) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def delete_by_id(elf, schema_name: str, id: str) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def delete_many(self, schema_name: str, **filters) -> any:
        """ """
        raise NotImplementedError

    @abstractmethod
    def count(self, schema_name: str, **filters) -> any:
        """ """
        raise NotImplementedError

Subclasses

Class variables

var db_url

Methods

def count(self, schema_name: str, **filters) ‑> 
Expand source code
@abstractmethod
def count(self, schema_name: str, **filters) -> any:
    """ """
    raise NotImplementedError
def delete(self, schema_name: str, filters: dict, first: bool = False) ‑> 
Expand source code
@abstractmethod
def delete(self, schema_name: str, filters: dict, first: bool = False) -> any:
    """ """
    raise NotImplementedError
def delete_by_id(elf, schema_name: str, id: str) ‑> 
Expand source code
@abstractmethod
def delete_by_id(elf, schema_name: str, id: str) -> any:
    """ """
    raise NotImplementedError
def delete_many(self, schema_name: str, **filters) ‑> 
Expand source code
@abstractmethod
def delete_many(self, schema_name: str, **filters) -> any:
    """ """
    raise NotImplementedError
def delete_one(self, schema_name: str, **filters) ‑> 
Expand source code
@abstractmethod
def delete_one(self, schema_name: str, **filters) -> any:
    """ """
    raise NotImplementedError
def fetch(self, schema_name: str, id: str = None, first: bool = False, limit: int = None, skip: int = None, **filters) ‑> List[dict]
Expand source code
@abstractmethod
def fetch(self, schema_name: str, id: str = None, first: bool = False, limit: int = None, skip: int = None,
          **filters) -> List[dict]:
    """             
    """
    raise NotImplementedError
def fetch_by_id(self, schema_name: str, id: str) ‑> dict

Get first item match by id

Expand source code
@abstractmethod
def fetch_by_id(self, schema_name: str, id: str) -> dict:
    """ Get first item match by id """
    return self.fetch(schema_name, first=True, id=id)
def fetch_many(self, schema_name: str, **filters) ‑> List[dict]

Get all items that matched

Expand source code
@abstractmethod
def fetch_many(self, schema_name: str, **filters) -> List[dict]:
    """ Get all items that matched  """
    return self.fetch(schema_name, **filters)

    # Inserting new data
def fetch_one(self, schema_name: str, **filters) ‑> dict

Get first item match

Expand source code
@abstractmethod
def fetch_one(self, schema_name: str, **filters) -> dict:
    """ Get first item match """
    return self.fetch(schema_name, first=True, **filters)
def insert(self, schema: marshmallow.schema.Schema, data: Union[dict, List[dict]]) ‑> 
Expand source code
@abstractmethod
def insert(self, schema: Schema, data: Union[dict, List[dict]]) -> any:
    """ """
    raise NotImplementedError
def insert_many(self, schema: marshmallow.schema.Schema, data: List[dict], validate_percentage: float = 1.0) ‑> 

TODO validate_percentage: what percentage of the list of data provided should be validated (0.5 is 50%)

Expand source code
@abstractmethod
def insert_many(self, schema: Schema, data: List[dict], validate_percentage: float = 1.0) -> any:
    """
        TODO `validate_percentage`: what percentage of the list of data provided should be validated (0.5 is 50%)
    """
    return self.insert(schema, data)
def insert_one(self, schema: marshmallow.schema.Schema, data: dict = None, **data_kwargs) ‑> 

Insert dict to db

Expand source code
@abstractmethod
def insert_one(self, schema: Schema, data: dict = None, **data_kwargs) -> any:
    """ Insert dict to db """
    return self.insert(schema, data or data_kwargs)
def insert_with_id(self, schema: marshmallow.schema.Schema, id: Union[str, int], data: dict = None, **data_kwargs) ‑> 

Insert dict with a specific 'id'

Expand source code
@abstractmethod
def insert_with_id(self, schema: Schema, id: Union[str, int], data: dict = None, **data_kwargs) -> any:
    """ Insert dict with a specific 'id' """
    raise NotImplementedError
def update(self, schema_name: str, filters: dict, data: Union[dict, List[dict]], first: bool = False) ‑> 
Expand source code
@abstractmethod
def update(self, schema_name: str, filters: dict, data: Union[dict, List[dict]], first: bool = False) -> any:
    """ """
    raise NotImplementedError
def update_many(self, schema_name: str, filters: dict, data: List[dict], validate_percentage: float = 1.0) ‑> 

validate_percentage: what percentage of the list of data provided should be validated (0.5 is 50% of items in list)

Expand source code
@abstractmethod
def update_many(self, schema_name: str, filters: dict, data: List[dict], validate_percentage: float = 1.0) -> any:
    """
        `validate_percentage`: what percentage of the list of data provided should be validated (0.5 is 50% of items in list)
    """
    raise NotImplementedError
def update_on_id(self, schema_name: str, id: Union[str, int], data: dict) ‑> 
Expand source code
@abstractmethod
def update_on_id(self, schema_name: str, id: Union[str, int], data: dict) -> any:
    """ """
    raise NotImplementedError
def update_one(self, filters: dict, data: dict, append: bool = False, schema: marshmallow.schema.Schema = None) ‑> dict
Expand source code
@abstractmethod
def update_one(self, filters: dict, data: dict, append: bool = False, schema: Schema = None) -> dict:
    """ """
    raise NotImplementedError