[WIP] pyDKB/storages: make ES client aware of storage scheme.

Author: mgolosova

Utils/Dataflow/pyDKB/storages/client/es.py

@@ -6,8 +6,13 @@
 from pyDKB.common.misc import try_to_import
 from pyDKB.common.types import logLevel
 
+from ..exceptions import (NotFound, InvalidRequest)
+
 
 _ESClient = try_to_import('elasticsearch', 'Elasticsearch')
+_NotFoundError = try_to_import('elasticsearch.exceptions', 'NotFoundError')
+_RequestError = try_to_import('elasticsearch.exceptions', 'RequestError')
+
 
 ParentClientClass = _ESClient if _ESClient else object
 
@@ -16,6 +21,7 @@ class ESClient(Client, ParentClientClass):
     """ Implement common interface for ES client. """
 
     index = None
+    scheme_info = None
 
     def __init__(self, *args, **kwargs):
         """ Initialize instance as parent client class object. """
@@ -71,5 +77,129 @@ def configure(self, cfg):
         if cfg.get('index'):
             self.index = cfg['index']
 
+        if cfg.get('scheme_info'):
+            self.scheme_info = cfg['scheme_info']
+
         # Re-initialize self as parent client class instance
         ParentClientClass.__init__(self, **kwargs)
+
+    def set_index(self, index):
+       """ Set default index name.
+
+       :param index: default index name
+       :type index: str
+       """
+       if index:
+           self.index = index
+
+    def set_scheme(self, scheme):
+       """ Set default index name.
+
+       Schema info should be provided in the following form:
+       {
+         'index': {INDEX_NAME: {"scheme": SCHEMA_NAME}, ...},
+         'scheme': {SCHEMA_NAME: {
+             "default_type": DEFAULT_DOC_TYPE,
+             DOC_TYPE_1: {},
+             DOC_TYPE_2: {"parent_type": DOC_TYPE_N},
+             ...
+           },
+           ...
+         }
+       }
+
+       :param scheme: storage scheme
+       :type scheme: dict
+       """
+       if scheme:
+           self.scheme_info = scheme
+
+    def is_child_type(self, doc_type, index=None, scheme=None):
+       """ Check if given document type is a child type.
+
+       One of additional parameters (`index`, `scheme`) must be specified.
+       If both specified, `index` is ignored.
+
+       :param doc_type: type name
+       :type doc_type: str
+       :param index: index name for which given type is checked
+       :type index: str
+       :param scheme: scheme name for which given type is checked
+       :type scheme: str
+
+       :return: True if type is a child type, False otherwise (or not enough
+                information about scheme provided)
+       :rtype: bool
+       """
+       scheme_info = self.scheme_info
+       if not (scheme_info and isinstance(scheme_info, dict)):
+           return False
+       if not index:
+           index = self.index
+       if not scheme and index:
+           idx_scheme = scheme_info.get('index', {}).get(index, {})\
+                        .get('scheme')
+           scheme = idx_scheme
+       if scheme:
+           type_scheme = scheme_info.get('scheme', {}).get(scheme, {})\
+                         .get(doc_type, {})
+           if type_scheme.get('parent_type'):
+               return True
+       return False
+
+    def get_default_type(self, index):
+       """ Get default document type for given index.
+
+       :param index: index name
+       :type index: str
+
+       :return: type name or None if information not available
+       :rtype: str, NoneType
+       """
+       scheme_info = self.scheme_info
+       if not (scheme_info and isinstance(scheme_info, dict)):
+           return False
+       scheme_name = scheme_info.get('index', {}).get(index, {})\
+                     .get('scheme')
+       doc_type = scheme_info.get('scheme', {}).get(scheme_name, {})\
+                  .get('default_type')
+       return doc_type
+
+    def get(self, oid, fields=None, **kwargs):
+       """ Get document from index by ID.
+
+       :param oid: document identifier
+       :type oid: str, int
+       :param fields: document fields to retrieve. If not specified
+                      or None passed, all available fields will be
+                      retrieved
+       :type fields: list(str)
+       :param index: index to get document from. If not specified,
+                     default index will be used (if configured).
+       :type index: str
+       :param *kwargs: see ``Elasticsearch.get()`` description
+
+       :return: record with given ID
+       :rtype: dict
+       """
+       # If index is not passed, default one should be used
+       index = kwargs.pop('index', self.index)
+       if not index:
+           raise InvalidRequest("Index not specified.")
+       doc_type = kwargs.pop('doc_type', self.get_default_type(index))
+       if not doc_type:
+           raise InvalidRequest("Document type not specified.")
+       if fields is not None:
+           kwargs['_source'] = fields
+       try:
+           r = ParentClientClass.get(self, id=oid, index=index, doc_type=doc_type,
+                                     **kwargs)
+       except _NotFoundError, err:
+           raise NotFound(id=oid, index=index, doc_type=doc_type, **kwargs)
+       except _RequestError, err:
+           if self.is_child_type(doc_type, index=index) \
+                   and err.args[1] == 'routing_missing_exception':
+               self.log('Parent info missed.', logLevel.WARN)
+               raise NotFound(id=oid, index=index, doc_type=doc_type, **kwargs)
+           raise InvalidRequest(err)
+       return r.get('_source', {})

Utils/Dataflow/pyDKB/storages/exceptions.py

@@ -23,3 +23,28 @@ def __init__(self, **kwargs):
             params = ', '.join(params)
             message = message + ' (%s)' % params
         super(NotFound, self).__init__(message)
+
+
+class InvalidRequest(StorageException):
+    """ Exception indicating wrong user request. """
+
+    def __init__(self, message, *args, **kwargs):
+        """ Initialize exception.
+
+        Message formatting: old-style ('%' operator) only.
+
+        :param message: error message
+        :type message: str
+        :param args: message format positional parameters
+        :type args: list
+        :param kwargs: message format named parameters
+        :type kwargs: dict
+        """
+        if args and kwargs:
+            raise ValueError("Message formatting supports only one type "
+                             "of parameters: positional OR named.")
+        if args:
+            message = message % params
+        elif kwargs:
+            message = message % kwargs
+        super(InvalidRequest, self).__init__(message)