# -*- coding: iso-8859-1 -*-
#-----------------------------------------------------------------------------
# Modeling Framework: an Object-Relational Bridge for python
#
# Copyright (c) 2001-2004 Sbastien Bigaret <sbigaret@users.sourceforge.net>
# All rights reserved.
#
# This file is part of the Modeling Framework.
#
# This code is distributed under a "3-clause BSD"-style license;
# see the LICENSE file for details.
#-----------------------------------------------------------------------------
"""A ModelSet holds a set of models that can co-exist and/or co-operate at runtime.
The framework'core can only handle one ModelSet at a time: the one returned by
`defaultModelSet()`. The module & class main responsabilities are:
1. making sure that all the entities defined in a ModelSet instance have
distinct names: a ModelSet identifies its models and entities by name.
Hence, object defaultModelSet() can be asked at runtime for models and
entities, given their names
2. registering the object returned by defaultModelSet() as the default
receiver for the notification
`ClassDescriptionNeededForEntityNameNotification` posted by
`ClassDescription.classDescriptionForName()`.
CVS informations:
$Id: ModelSet.py 963 2005-05-03 15:12:06Z sbigaret $
"""
__version__='$Revision: 963 $'[11:-2]
from Modeling.utils import isaValidName,base_persistent_object
from Modeling.Model import ModelError,Model,loadXMLModel
from Modeling.ClassDescription import ClassDescriptionNeededForEntityNameNotification
from NotificationFramework import NotificationCenter
NC=NotificationCenter
from Modeling import ClassDescription
import types
from Modeling.logging import error,warn
from threading import RLock
defaultModelSet_lock=RLock()
lock=defaultModelSet_lock.acquire
unlock=defaultModelSet_lock.release
__defaultModelSet=None
def defaultModelSet():
"""
Returns the model set to use globally in an application
There can be only one such ModelSet.
"""
lock()
global __defaultModelSet
try:
if __defaultModelSet is None:
__defaultModelSet=ModelSet()
return __defaultModelSet
finally:
unlock()
defaultModelGroup=defaultModelSet
def setDefaultModelSet(aModelSet):
"""
Sets the model set to use globally in an application
There can be only one such ModelSet.
"""
lock()
global __defaultModelSet
try:
__defaultModelSet=ModelSet()
finally:
unlock()
setDefaultModelGroup=setDefaultModelSet
def updateModelWithCFG(model, cfg_path):
"""
Deprecated: use Model.updateModelWithCFG instead. This method will be
removed in v0.9.1
"""
from utils import deprecated
deprecated('ModelSet.updateModelWithCFG','Model.updateModelWithCFG()',
'0.9.1')
from Modeling import Model
return Model.updateModelWithCFG(model, cfg_path)
class ModelSet(base_persistent_object):
"""Holds a set of Modeling.Models that can co-exist at runtime
Two models *cannot* be stored within a single ModelSet if and only if there
is an entity in the first model that has the same name than a entity in the
second model.
"""
isLogEnabled=0
def __init__(self):
"Initializes a modelset"
self._models=()
self._name=''
def addModel(self, aModel):
"""
Add a model to the model set. In order to be successfully added the model
should not have the same name than one of the models already added to the
ModelSet, nor should it hold an entity whose name is already used within
the ModelSet.
Raises ModelError if the insertion of aModel fails.
If the environment variable 'MDL_DB_CONNECTIONS_CFG' is set, the file
it points to is used to update aModel's connection dictionary (and
possibly its adaptorName as well). See Model.updateModelWithCFG() for
details.
"""
#assert
# Check model name
if aModel.name() in self.modelsNames():
raise ModelError, "Attempt to insert a model in modelset '%s' but a model named '%s' is already registered." % (self.name(), aModel.name())
# Entities share a single namespace: check this!
my_entities_names=self.entitiesNames()
for _entity in aModel.entities():
if _entity.name() in my_entities_names:
raise ModelError, "Attempt to insert a model in modelset '%s' containing an entity named '%s': that name is already registered by an other entity" % (self.name(), _entity.name())
_models=list(self.models())
_models.append(aModel)
self._models=tuple(_models)
aModel._setModelSet(self)
# MDL_DB_CONNECTIONS_CFG
import os
cfg_path=os.environ.get('MDL_DB_CONNECTIONS_CFG')
if cfg_path:
from Model import updateModelWithCFG
updateModelWithCFG(aModel, cfg_path)
aModel.cacheSimpleMethods()
# XML Import/Export facilities
def addModelFromXML(self, xmlSource):
"""
Instanciates the model from datas in xmlSource and adds it to the modelSet.
Parameter 'xmlSource' is a dictionary containing the key 'string' or 'file'
with the corresponding value. If both keys are provided, the result is
undefined.
Example: aModelSet.addModelFromXML({'file': '/home/user/model.xml'})
Returns: the brand new model
"""
#import pdb; pdb.set_trace()
if xmlSource.has_key('string'):
xml_content=xmlSource['string']
elif xmlSource.has_key('file'):
f=open(xmlSource['file'], 'rb')
xml_content=f.read()
f.close()
else:
raise AttributeError, "xmlSource parameter has no key 'string' or 'file'"
model=loadXMLModel(xml_content)
self.addModel(model)
return model
def classDescriptionForEntityName(self, entityName):
"""
Returns the class description for that entity, or None if that entity
cannot be found.
You should rarely need to call this method directly ; instead, use the
ClassDescription's static method 'classDescriptionForName()' which takes
care of finding the appropriate class description.
See also: Entity.classDescriptionForInstances()
ClassDescription.classDescriptionForName()
"""
entity=self.entityNamed(entityName)
if entity:
return entity.classDescriptionForInstances()
return None
# shortcut + standard API?
classDescriptionForName=classDescriptionForEntityName
def getXMLDOMForModelNamed(self, aModelName, encoding='iso-8859-1'):
"Returns the XML DOM for the specified model, or None if it does not exist"
_model=self.modelNamed(aModelName)
if not _model: return None
return _model.getXMLDOM(doc=None, parentNode=None, encoding=encoding)
def getXMLStreamForModelNamed(self, aModelName, encoding='iso-8859-1'):
"""
Returns the XML stream for the specified model, or None if it does not
exist.
"""
xmldom=self.XMLDOMForModelNamed(self, aModelName, encoding)
return
def entitiesNames(self):
"""
Returns the whole set of entities' name controlled by the receiver.
Returns: a tuple
"""
_list=[]
for _model in self._models:
_list=_list+list(_model.entitiesNames())
return tuple(_list)
def entityNamed(self, aName):
"""
Search an entity named 'aName' in the models contained in this modelset,
and returns it, or None if it cannot be found
"""
for _model in self._models:
if _model.entityNamed(aName) is not None:
return _model.entityNamed(aName)
return None
def modelForEntityNamed(self, aName):
"""
Returns the model holding the entity named 'aName', or 'None' if
it cannot be found
"""
for _model in self._models:
if _model.entityNamed(aName) is not None:
return _model
return None
def modelNamed(self, aName):
"Returns the model named 'aName', or None if it cannot be found"
for _model in self.models():
if _model.name() == aName: return _model
return None
def modelsNames(self):
"Returns the full set of contained models'names"
return map(lambda o: o.name(), self._models)
def models(self):
"Returns the full set of contained models"
return self._models
def name(self):
"Returns the modelsSet's name"
return self._name
def removeModel(self, aModel):
"""
Removes the supplied model from the registered models.
Parameter 'aModel' can be either an 'Model' instance (or a instance of a
class deriving from Model) or a model's name.
No referential checks are done.
Raises 'ModelError' if 'aModel' cannot be found.item
"""
#assert
if type(aModel)!=types.StringType:
aModelName = aModel.name()
else:
aModelName = aModel
_model=self.modelNamed(aModelName)
if _model is None:
raise ModelError, "ModelSet '%s' has no modelcorresponding to parameter a'Model' ('%s')" % (self.name(), str(aModel))
try:
_models = list(self._models)
_models.remove(_model)
self._models=tuple(_models)
except ValueError:
# pb: None has no name()!!!
raise ModelError, "ModelSet '%s' has no model named '%s'" % (self.name(), _model.name())
_model._setModelSet(None)
def setName(self, aName):
"Sets the modelsSet's name"
self._name=aName
def handleNotification(self, notification):
"""
Handles notification
'ClassDescription.ClassDescriptionNeededForEntityNameNotification'.
"""
if notification.name() == ClassDescriptionNeededForEntityNameNotification:
name=notification.object()
cd=self.classDescriptionForEntityName(name)
if cd:
ClassDescription.registerClassDescription(cd, name)
else:
error('ModelSet got an unhandled notification: %s'%notification.name())
##
## Register the default ModelSet as a listener for notification
## ClassDescriptionNeededForEntityNameNotification
##
#def __registerModelSet():
import code
_code=code.compile_command('from Modeling import ModelSet;'\
'observer_object=ModelSet.defaultModelSet()')
NC.addObserver(_code,
ModelSet.handleNotification,
ClassDescriptionNeededForEntityNameNotification )
del _code
#__registerModelSet()
|