# **************************************************************************
# *
# * Authors: J.M. De la Rosa Trevin (jmdelarosa@cnb.csic.es)
# *
# * 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
# * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# * 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 'scipion@cnb.csic.es'
# *
# **************************************************************************
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