Source code for pyworkflow.mapper.mapper

# **************************************************************************
# *
# * Authors:     J.M. De la Rosa Trevin (
# *
# * Unidad de  Bioinformatica of Centro Nacional de Biotecnologia , CSIC
# *
# * This program is free software; you can redistribute it and/or modify
# * it under the terms of the GNU General Public License as published by
# * the Free Software Foundation; either version 3 of the License, or
# * (at your option) any later version.
# *
# * This program is distributed in the hope that it will be useful,
# * but WITHOUT ANY WARRANTY; without even the implied warranty of
# * GNU General Public License for more details.
# *
# * You should have received a copy of the GNU General Public License
# * along with this program; if not, write to the Free Software
# * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
# * 02111-1307  USA
# *
# *  All comments concerning this program package may be sent to the
# *  e-mail address ''
# *
# **************************************************************************
import logging
logger = logging.getLogger(__name__)

import pyworkflow.object as obj

[docs]class Mapper: """This class will serves as a Data Mapper pattern. It will store/retrieve objects from some storage environment. (like SQL, XML or others) The mapper should have access to class dictionary in order to build any give class by name""" # Just to avoid print several times the same warning __warnings = set() ORIGINAL_CLASS_NAME_ATTRIBUTE = "oldClassName"
[docs] @staticmethod def annotateClassName(instance, oldClassName): """ Annotate an object with the original class name""" setattr(instance, Mapper.ORIGINAL_CLASS_NAME_ATTRIBUTE, oldClassName)
[docs] @staticmethod def getObjectPersistingClassName(instance): if hasattr(instance, Mapper.ORIGINAL_CLASS_NAME_ATTRIBUTE): return getattr(instance, Mapper.ORIGINAL_CLASS_NAME_ATTRIBUTE) else: return instance.getClassName()
def __init__(self, dictClasses=None): if dictClasses: self.dictClasses = dictClasses else: self.dictClasses = dir(obj)
[docs] def warning(self, msg): if msg not in self.__warnings: print("WARNING: %s" % msg) self.__warnings.add(msg)
def _buildObjectFromClass(self, className, **kwargs): """ Build an instance of an object given the class name, it should be in the classes dictionary. """ if className not in self.dictClasses: self.warning("Class '%s' not found in mapper dict. Ignored. " % className) return None objClass = self.dictClasses[className] if not issubclass(objClass, obj.Object): print("WARNING: Class '%s' is not a subclass of Object. Ignored. " % className) return None try: instance = self.dictClasses[className](**kwargs) except Exception as e: clazz = self.dictClasses._default logger.error('Class %s could not be created. Replacing it with %s ' % (className, clazz.__name__), exc_info=e) instance = clazz(**kwargs) # If it's the default class in the dictionary if objClass.__name__ != className: # ... annotate original class name self.annotateClassName(instance, className) return instance def _getStrValue(self, value): """ Return empty string if value is None or empty. """ if not value: return '' return value
[docs] def commit(self): """Commit changes made to the storage""" pass
[docs] def insert(self, obj): """Insert a new object into the system, the id will be set""" pass
[docs] def update(self, obj, direction='to'): """Update an existing object, the id should not be None direction can be "to" or "from" which indicates the priority of the update. If "to" is used, object data is put in storage. If "from", object data is retrieved from storage""" if direction == 'to': # Update db with object changes self.updateTo(obj) elif direction == 'from': # Update object with db info self.updateFrom(obj) else: raise Exception('Invalid option %s for Sqlite.updateObject' % direction)
[docs] def updateFrom(self, obj): """Update object data with storage info""" pass
[docs] def updateTo(self, obj): """Update storage with object info""" pass
[docs] def store(self, obj): """Stores an object, it can be inserted or updated""" if obj._objId is None: self.insert(obj) else: self.updateTo(obj)
[docs] def selectById(self, objId): """Return the object which id is objId""" pass
[docs] def exists(self, objId): """Return True if the id is in the database""" pass
[docs] def selectAll(self, iterate=False): """Return all object from storage""" pass
[docs] def selectFirst(self): """Return only the first element""" for obj in self.selectAll(iterate=True): return obj return None
[docs] def selectBy(self, **args): """Select object meetings some criteria""" pass
[docs] def selectByClass(self, className, includeSubclasses=True): """Select all object of a give class. By default object of subclasses will be retrieved also. This behaviour can be changed by passing includeSubclass=False""" return self.selectBy(classname=className)
[docs] def getParent(self, obj): """ Retrieve the parent object of another. """ pass
[docs] def getFullName(self, obj): """ Return the full object name such as: grandparent.parent.objName """ nameParts = [] parent = obj while parent: nameParts.insert(0, parent.getLastName()) parent = self.getParent(parent) return '.'.join(nameParts)
[docs] def insertRelation(self, relName, creatorObj, parentObj, childObj): """ This function will add a new relation between two objects. Params: relName: the name of the relation to be added. creatorObj: this object will be the one who register the relation. parentObj: this is "parent" in the relation childObj: this is "child" in the relation """ pass
[docs] def getRelationChilds(self, relName, parentObj): """ Return all "child" objects for a given relation. Params: relName: the name of the relation. parentObj: this is "parent" in the relation Returns: a list of "child" objects. """ pass
[docs] def getRelationParents(self, relName, childObj): """ Return all "parent" objects for a given relation. Params: relName: the name of the relation. childObj: this is "child" in the relation Returns: a list of "parent" objects. """ pass
[docs] def getRelationsByCreator(self, creatorObj): """ Return all relations created by creatorObj. """ pass
[docs] def deleteRelations(self, creatorObj): """ Delete all relations created by object creatorObj """ pass