Source code for ripozo.managers.base

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

from abc import ABCMeta, abstractmethod, abstractproperty

from ripozo.decorators import classproperty

import logging
import six
logger = logging.getLogger(__name__)


[docs]@six.add_metaclass(ABCMeta) class BaseManager(object): """ The BaseManager implements some common methods that are valuable across all databases as well as expose an interface that must be implemented by any specific implementation. This needs to be extended in order to implement a new database type. This should handle all direct interactions with a database or ORM. The extended classes are injected into viewsets in order to specify how to get the data from the database. :param unicode pagination_pk_query_arg: The name of the query parameter that specifies the page or pk for the next set page to return when paginating over a list :param unicode pagination_count_query_arg: The name of the query parameter that specifies the maximum number of results to return in a list retrieval :param unicode pagination_next: The meta parameter to return that specifies the next query parameters :param int paginate_by: The number of results to return by default. This gets overridden by pagination_count_query_arg :param list order_by: A list of the fields to order the results by. This may be restricted in certain databases :param list fields: A list of the fields that are able to be manipulated or retrieved by the manager :param type model: The model that is being managed. This is the individual model that is set by the user. For any type of base class this should be None. However, it is required for actual implementations """ pagination_pk_query_arg = 'pagination_pk' pagination_count_query_arg = 'count' pagination_next = 'next' paginate_by = 10000 order_by = None _fields = None _list_fields = None model = None arg_parser = None _field_validators = None
[docs] @abstractmethod def create(self, values, *args, **kwargs): """ Create a model with the values according to the values dictionary :param values: A dictionary of values to create the model according to :type values: dict :param args: Extra arguments :type args: tuple :param kwargs: Extra keyword arguments :type kwargs: dict :return: The dictionary of arguments that should be returned by the serializer :rtype: dict """ pass
[docs] @abstractmethod def retrieve(self, lookup_keys, *args, **kwargs): """ Retrieve a single model and nothing more as a python dictionary :param lookup_keys: The lookup keys for the model and the associated values :type lookup_keys: dict :param args: Extra arguments :type args: tuple :param kwargs: Extra keyword arguments :type kwargs: dict :return: The dictionary of arguments that should be returned by the serializer :return: A tuple with the first object being a dictionary of key value pairs according to the fields list and the second object is the meta data (such as the next url for a paginated list) :rtype: tuple """ pass
[docs] @abstractmethod def retrieve_list(self, filters, *args, **kwargs): """ Retrieves a list of dictionaries containing the fields for the associated model :param lookup_keys: The lookup keys for the model and the associated values :type lookup_keys: dict :param args: Extra arguments :type args: tuple :param kwargs: Extra keyword arguments :type kwargs: dict :return: The dictionary of arguments that should be returned by the serializer :return: A a list of dictionaries of key value pairs according to the fields list :rtype: list """ pass
[docs] @abstractmethod def update(self, lookup_keys, updates, *args, **kwargs): """ Updates the model found with the lookup keys according to the updates dictionary where the keys are the fields to update and the values are the new values for the field :param lookup_keys: The keys to find the object that is to be updated :type lookup_keys: dict :param updates: The fields to update and their associated new update values :type updates: dict :param args: Extra arguments :type args: tuple :param kwargs: Extra keyword arguments :type kwargs: dict :return: A dictionary of the full updated model according to the fields class attribute :rtype: dict """ pass
[docs] @abstractmethod def delete(self, lookup_keys, *args, **kwargs): """ Deletes the model found with the lookup keys :param lookup_keys: The keys with which to find the model to delete :type lookup_keys: dict :param args: Extra arguments :type args: tuple :param kwargs: Extra keyword arguments :type kwargs: dict :return: nothing. :rtype: NoneType """ pass
[docs] @abstractmethod def get_field_type(self, name): """ Returns the BaseField instance (or subclass) instance that corresponds to the named attribute on the model. For example, if you the column "name" on the model for this manager is a String column, then this should return an instance of StringField with the appropriate requirements. :param name: The name of the field on the model for which you are getting the field name :type name: unicode :return: The BaseField class (or subclass) instance to handle the field specified by the name :rtype: ripozo.viewsets.fields.base.BaseField """ pass
@classproperty def fields(cls): """ Simply makes sure that the _fields attribute is not None. Returns [] if cls._fields evaluates to None or some equivalent. """ return cls._fields or [] @classproperty def list_fields(cls): """ These are the fields that should be used for retrieving a list of models. This is often necessary for performance reasons when you only want the ids to create links to the individual resource and not the full resource :return: The list fields, if the ``cls._list_fields`` attribute is set, otherwise, ``cls.fields`` :rtype: list """ return cls._list_fields or cls.fields @classproperty def field_validators(cls): """ Gets the BaseField instances for all of the fields on the manager. :return: :rtype: list """ cls._field_validators = cls._field_validators or {} for field_name in cls.fields: if field_name not in cls._field_validators: cls._field_validators[field_name] = (cls.get_field_type(field_name)) return list(cls._field_validators.values()) @abstractproperty def queryset(self): """ Gets the list of models that are valid to operate on :return: A list of models :rtype: list """ pass
[docs] def get_pagination_count(self, filters): """ Get the pagination count from the args :param filters: All of the args :type filters: dict :return: tuple of (pagination_count, updated_filters :rtype: tuple """ # get the pagination count or else use the default filters = filters.copy() pagination_count_query = filters.pop(self.pagination_count_query_arg, None) pagination_count = int(pagination_count_query) if pagination_count_query else self.paginate_by logger.debug('Paginating list by {0}'.format(pagination_count)) return pagination_count, filters
[docs] def get_pagination_pks(self, filters): """ Get the pagination pks from the args :param filters: All of the args :type filters: dict :return: tuple of (pagination_pks, updated_filters :rtype: tuple """ filters = filters.copy() last_pagination_pk = filters.pop(self.pagination_pk_query_arg, None) return last_pagination_pk, filters
[docs] def dot_field_list_to_dict(self, fields=None): """ Converts a list of dot delimited fields (and related fields) and turns it into a dictionary for example, it would transform .. code-block:: python >>> dot_field_list_to_dict(['id', 'value', 'related.id', 'related.related_value']) { 'id': None, 'value': None, 'related': { 'id': None, 'related_value': None } } :param list fields: :return: A dictionary of the fields layered as to indicate relationships. :rtype: dict """ # TODO find a better fucking way field_dict = {} fields = fields or self.fields for f in fields: field_parts = f.split('.') current = field_dict part = field_parts.pop(0) while len(field_parts) > 0: current[part] = current.get(part, dict()) current = current[part] part = field_parts.pop(0) current[part] = None return field_dict