appypod-rattail/gen/plone25/mixins/__init__.py

1277 lines
59 KiB
Python

'''This package contains mixin classes that are mixed in with generated classes:
- mixins/ClassMixin is mixed in with Standard Archetypes classes;
- mixins/ToolMixin is mixed in with the generated application Tool class;
- mixins/FlavourMixin is mixed in with the generated application Flavour
class.
The AbstractMixin defined hereafter is the base class of any mixin.'''
# ------------------------------------------------------------------------------
import os, os.path, sys, types, mimetypes
import appy.gen
from appy.gen import String, Selection
from appy.gen.utils import FieldDescr, GroupDescr, PhaseDescr, StateDescr, \
ValidationErrors, sequenceTypes, SomeObjects
from appy.gen.plone25.descriptors import ArchetypesClassDescriptor
from appy.gen.plone25.utils import updateRolesForPermission, getAppyRequest
# ------------------------------------------------------------------------------
class AbstractMixin:
'''Every Archetype class generated by appy.gen inherits from a mixin that
inherits from this class. It contains basic functions allowing to
minimize the amount of generated code.'''
def createOrUpdate(self, created):
'''This method creates (if p_created is True) or updates an object.
In the case of an object creation, p_self is a temporary object
created in the request by portal_factory, and this method creates
the corresponding final object. In the case of an update, this
method simply updates fields of p_self.'''
rq = self.REQUEST
obj = self
if created:
obj = self.portal_factory.doCreate(self, self.id) # portal_factory
# creates the final object from the temp object.
previousData = None
if not created: previousData = self.rememberPreviousData()
# Perform the change on the object, unless self is a tool being created.
if (obj._appy_meta_type == 'tool') and created:
# We do not process form data (=real update on the object) if the
# tool itself is being created.
pass
else:
obj.processForm()
if previousData:
# Keep in history potential changes on historized fields
self.historizeData(previousData)
# Manage references
obj._appy_manageRefs(created)
if obj.wrapperClass:
# Get the wrapper first
appyWrapper = obj._appy_getWrapper(force=True)
# Call the custom "onEdit" if available
try:
appyWrapper.onEdit(created)
except AttributeError, ae:
pass
# Manage "add" permissions
obj._appy_managePermissions()
# Reindex object
obj.reindexObject()
return obj
def delete(self):
'''This methods is self's suicide.'''
self.getParentNode().manage_delObjects([self.id])
def onCreate(self):
'''This method is called when a user wants to create a root object in
the application folder or an object through a reference field.'''
rq = self.REQUEST
if rq.get('initiator', None):
# The object to create will be linked to an initiator object through
# a ref field.
initiatorRes=self.uid_catalog.searchResults(UID=rq.get('initiator'))
rq.SESSION['initiator'] = rq.get('initiator')
rq.SESSION['initiatorField'] = rq.get('field')
rq.SESSION['initiatorTarget'] = rq.get('type_name')
if self._appy_meta_type == 'tool':
if rq.get('initiator', None):
# This is the creation of an object linked to the tool
baseUrl = self.absolute_url()
else:
# This is the creation of a root object in the app folder
baseUrl = self.getAppFolder().absolute_url()
else:
baseUrl = self.absolute_url()
objId = self.generateUniqueId(rq.get('type_name'))
urlBack = '%s/portal_factory/%s/%s/skyn/edit' % \
(baseUrl, rq.get('type_name'), objId)
return self.goto(urlBack)
def onUpdate(self):
'''This method is executed when a user wants to update an object.
The object may be a temporary object created by portal_factory in
the request. In this case, the update consists in the creation of
the "final" object in the database. If the object is not a temporary
one, this method updates its fields in the database.'''
rq = self.REQUEST
# Dict for storing validation errors
errors = {}
errorMessage = self.translate(
'Please correct the indicated errors.', domain='plone')
# Go back to the consult view if the user clicked on 'Cancel'
if rq.get('buttonCancel', None):
if '/portal_factory/' in self.absolute_url():
# Go back to the Plone site (no better solution at present).
urlBack = self.portal_url.getPortalObject().absolute_url()
else:
urlBack = '%s/skyn/view' % self.absolute_url()
self.plone_utils.addPortalMessage(
self.translate('Changes canceled.', domain='plone'))
return self.goto(urlBack, True)
# Trigger field-specific validation
self.validate(REQUEST=rq, errors=errors, data=1, metadata=0)
if errors:
rq.set('errors', errors)
self.plone_utils.addPortalMessage(errorMessage)
return self.skyn.edit(self)
else:
# Trigger inter-field validation
self.validateAllFields(rq, errors)
if errors:
rq.set('errors', errors)
self.plone_utils.addPortalMessage(errorMessage)
return self.skyn.edit(self)
else:
# Create or update the object in the database
obj = self.createOrUpdate(rq.get('is_new') == 'True')
# Redirect the user to the appropriate page
if rq.get('buttonOk', None):
# Go to the consult view for this object
obj.plone_utils.addPortalMessage(
obj.translate('Changes saved.', domain='plone'))
urlBack = '%s/skyn/view' % obj.absolute_url()
return self.goto(urlBack, True)
elif rq.get('buttonPrevious', None):
# Go to the edit view (previous page) for this object
rq.set('fieldset', rq.get('previousPage'))
return obj.skyn.edit(obj)
elif rq.get('buttonNext', None):
# Go to the edit view (next page) for this object
rq.set('fieldset', rq.get('nextPage'))
return obj.skyn.edit(obj)
def onDelete(self):
rq = self.REQUEST
msg = self.translate('delete_done')
self.delete()
self.plone_utils.addPortalMessage(msg)
self.goto(rq['HTTP_REFERER'], True)
def rememberPreviousData(self):
'''This method is called before updating an object and remembers, for
every historized field, the previous value. Result is a dict
~{s_fieldName: previousFieldValue}~'''
res = {}
for atField in self.Schema().filterFields(isMetadata=0):
fieldName = atField.getName()
appyType = self.getAppyType(fieldName, asDict=False)
if appyType and appyType.historized:
res[fieldName] = (getattr(self, fieldName),
atField.widget.label_msgid)
return res
def addDataChange(self, changes, labels=False):
'''This method allows to add "manually" a data change into the objet's
history. Indeed, data changes are "automatically" recorded only when
a HTTP form is uploaded, not if, in the code, a setter is called on
a field. The method is also called by the method historizeData below,
that performs "automatic" recording when a HTTP form is uploaded.'''
# Add to the p_changes dict the field labels if they are not present
if not labels:
for fieldName in changes.iterkeys():
appyType = self.getAppyType(fieldName)
changes[fieldName] = (changes[fieldName], appyType['label'])
# Create the event to record in the history
DateTime = self.getProductConfig().DateTime
state = self.portal_workflow.getInfoFor(self, 'review_state')
user = self.portal_membership.getAuthenticatedMember()
event = {'action': '_datachange_', 'changes': changes,
'review_state': state, 'actor': user.id,
'time': DateTime(), 'comments': ''}
# Add the event to the history
histKey = self.workflow_history.keys()[0]
self.workflow_history[histKey] += (event,)
def historizeData(self, previousData):
'''Records in the object history potential changes on historized fields.
p_previousData contains the values, before an update, of the
historized fields, while p_self already contains the (potentially)
modified values.'''
# Remove from previousData all values that were not changed
for fieldName in previousData.keys():
prev = previousData[fieldName][0]
curr = getattr(self, fieldName)
if (prev == curr) or ((prev == None) and (curr == '')) or \
((prev == '') and (curr == None)):
del previousData[fieldName]
if previousData:
self.addDataChange(previousData, labels=True)
def goto(self, url, addParams=False):
'''Brings the user to some p_url after an action has been executed.'''
rq = self.REQUEST
if not addParams: return rq.RESPONSE.redirect(url)
# Add some context-related parameters if needed.
params = []
if rq.get('phase', ''): params.append('phase=%s' % rq['phase'])
if rq.get('pageName', ''): params.append('pageName=%s' % rq['pageName'])
if rq.get('nav', ''): params.append('nav=%s' % rq['nav'])
params = '&'.join(params)
if not params: return rq.RESPONSE.redirect(url)
if url.find('?') != -1: params = '&' + params
else: params = '?' + params
return rq.RESPONSE.redirect(url+params)
def getAppyValue(self, name, appyType=None, useParamValue=False,value=None):
'''Returns the value of field (or method) p_name for this object
(p_self). If p_appyType (the corresponding Appy type) is provided,
it gives additional information about the way to render the value.
If p_useParamValue is True, the method uses p_value instead of the
real field value (useful for rendering a value from the object
history, for example).'''
# Which value will we use ?
if useParamValue: v = value
else:
try:
v = eval('self.%s' % name)
except AttributeError:
# Probably a newly created attribute.
# In this case, return the default value.
v = None
if appyType: v = appyType['default']
if not appyType: return v
if (v == None) or (v == ''): return v
vType = appyType['type']
if vType == 'Date':
res = v.strftime('%d/%m/') + str(v.year())
if appyType['format'] == 0:
res += ' %s' % v.strftime('%H:%M')
return res
elif vType == 'String':
if not v: return v
if appyType['isSelect']:
validator = appyType['validator']
if isinstance(validator, Selection):
# Value(s) come from a dynamic vocabulary
return validator.getText(self, v)
else:
# Value(s) come from a fixed vocabulary whose texts are in
# i18n files.
maxMult = appyType['multiplicity'][1]
t = self.translate
if (maxMult == None) or (maxMult > 1):
return [t('%s_%s_list_%s' % (self.meta_type, name, e)) \
for e in v]
else:
return t('%s_%s_list_%s' % (self.meta_type, name, v))
return v
elif vType == 'Boolean':
if v: return self.translate('yes', domain='plone')
else: return self.translate('no', domain='plone')
elif vType == 'Float':
if appyType['precision'] == None:
v = str(v)
else:
format = '%%.%df' % appyType['precision']
v = format % v
return v
def getAppyType(self, fieldName, forward=True, asDict=True):
'''Returns the Appy type corresponding to p_fieldName. If you want to
get the Appy type corresponding to a backward field, set p_forward
to False and specify the corresponding Archetypes relationship in
p_fieldName.'''
res = None
if forward:
if fieldName == 'id': return res
if self.wrapperClass:
baseClass = self.wrapperClass.__bases__[-1]
try:
# If I get the attr on self instead of baseClass, I get the
# property field that is redefined at the wrapper level.
res = appyType = getattr(baseClass, fieldName)
if asDict:
res = self._appy_getTypeAsDict(
fieldName, appyType, baseClass)
except AttributeError:
# Check for another parent
if self.wrapperClass.__bases__[0].__bases__:
baseClass = self.wrapperClass.__bases__[0].__bases__[-1]
try:
res = appyType = getattr(baseClass, fieldName)
if asDict:
res = self._appy_getTypeAsDict(
fieldName, appyType, baseClass)
except AttributeError:
pass
else:
referers = self.getProductConfig().referers
for appyType, rel in referers[self.__class__.__name__]:
if rel == fieldName:
res = appyType
if asDict:
res = appyType.__dict__
res['backd'] = appyType.back.__dict__
return res
def _appy_getRefs(self, fieldName, ploneObjects=False,
noListIfSingleObj=False, startNumber=None):
'''p_fieldName is the name of a Ref field. This method returns an
ordered list containing the objects linked to p_self through this
field. If p_ploneObjects is True, the method returns the "true"
Plone objects instead of the Appy wrappers.
If p_startNumber is None, this method returns all referred objects.
If p_startNumber is a number, this method will return x objects,
starting at p_startNumber, x being appyType.maxPerPage.'''
appyType = self.getAppyType(fieldName)
sortedUids = getattr(self, '_appy_%s' % fieldName)
batchNeeded = startNumber != None
exec 'refUids= self.getRaw%s%s()' % (fieldName[0].upper(),fieldName[1:])
# There may be too much UIDs in sortedUids because these fields
# are not updated when objects are deleted. So we do it now. TODO: do
# such cleaning on object deletion?
toDelete = []
for uid in sortedUids:
if uid not in refUids:
toDelete.append(uid)
for uid in toDelete:
sortedUids.remove(uid)
# Prepare the result
res = SomeObjects()
res.totalNumber = res.batchSize = len(sortedUids)
if batchNeeded:
res.batchSize = appyType['maxPerPage']
if startNumber != None:
res.startNumber = startNumber
# Get the needed referred objects
i = res.startNumber
# Is it possible and more efficient to perform a single query in
# uid_catalog and get the result in the order of specified uids?
while i < (res.startNumber + res.batchSize):
if i >= res.totalNumber: break
refUid = sortedUids[i]
refObject = self.uid_catalog(UID=refUid)[0].getObject()
if not ploneObjects:
refObject = refObject.appy()
res.objects.append(refObject)
i += 1
if res.objects and noListIfSingleObj:
if appyType['multiplicity'][1] == 1:
res.objects = res.objects[0]
return res
def getAppyRefs(self, fieldName, forward=True, startNumber=None):
'''Gets the objects linked to me through p_fieldName. If you need to
get a backward reference, set p_forward to False and specify the
corresponding Archetypes relationship in p_fieldName.
If p_startNumber is None, this method returns all referred objects.
If p_startNumber is a number, this method will return x objects,
starting at p_startNumber, x being appyType.maxPerPage.'''
if forward:
return self._appy_getRefs(fieldName, ploneObjects=True,
startNumber=startNumber).__dict__
else:
# Note Pagination is not yet implemented for backward ref.
return SomeObjects(self.getBRefs(fieldName)).__dict__
def getAppyRefIndex(self, fieldName, obj):
'''Gets the position of p_obj within Ref field named p_fieldName.'''
sortedFieldName = '_appy_%s' % fieldName
sortedObjectsUids = getattr(self, sortedFieldName)
res = sortedObjectsUids.index(obj.UID())
return res
def getAppyBackRefs(self):
'''Returns the list of back references (=types, not objects) that are
defined for this class.'''
className = self.__class__.__name__
referers = self.getProductConfig().referers
res = []
if referers.has_key(className):
for appyType, relationship in referers[className]:
d = appyType.__dict__
d['backd'] = appyType.back.__dict__
res.append((d, relationship))
return res
def getAppyRefPortalType(self, fieldName):
'''Gets the portal type of objects linked to me through Ref field named
p_fieldName.'''
appyType = self.getAppyType(fieldName)
tool = self.getTool()
if self._appy_meta_type == 'flavour':
flavour = self._appy_getWrapper(force=True)
else:
portalTypeName = self._appy_getPortalType(self.REQUEST)
flavour = tool.getFlavour(portalTypeName)
return self._appy_getAtType(appyType['klass'], flavour)
def _appy_getOrderedFields(self, isEdit):
'''Gets all fields (normal fields, back references, fields to show,
fields to hide) in order, in the form of a list of FieldDescr
instances.'''
orderedFields = []
# Browse Archetypes fields
for atField in self.Schema().filterFields(isMetadata=0):
fieldName = atField.getName()
appyType = self.getAppyType(fieldName)
if not appyType:
if isEdit and (fieldName == 'title'):
# We must provide a dummy appy type for it. Else, it will
# not be rendered in the "edit" form.
appyType = String(multiplicity=(1,1)).__dict__
else:
continue # Special fields like 'id' are not relevant
# Do not display title on view page; it is already in the header
if not isEdit and (fieldName=='title'): pass
else:
orderedFields.append(FieldDescr(atField, appyType, None))
# Browse back references
for appyType, fieldRel in self.getAppyBackRefs():
orderedFields.append(FieldDescr(None, appyType, fieldRel))
# If some fields must be moved, do it now
res = []
for fieldDescr in orderedFields:
if fieldDescr.appyType['move']:
newPosition = len(res) - abs(fieldDescr.appyType['move'])
if newPosition <= 0:
newPosition = 0
res.insert(newPosition, fieldDescr)
else:
res.append(fieldDescr)
return res
def showField(self, fieldDescr, isEdit=False):
'''Must I show field corresponding to p_fieldDescr?'''
if isinstance(fieldDescr, FieldDescr):
fieldDescr = fieldDescr.__dict__
appyType = fieldDescr['appyType']
if isEdit and (appyType['type']=='Ref') and appyType['add']:return False
if isEdit and (appyType['type'] == 'Action'): return False
if (fieldDescr['widgetType'] == 'backField') and \
not self.getBRefs(fieldDescr['fieldRel']): return False
# Do not show field if it is optional and not selected in flavour
if appyType['optional']:
tool = self.getTool()
flavour = tool.getFlavour(self, appy=True)
flavourAttrName = 'optionalFieldsFor%s' % self.meta_type
flavourAttrValue = getattr(flavour, flavourAttrName, ())
if fieldDescr['atField'].getName() not in flavourAttrValue:
return False
# Check if the user has the permission to view or edit the field
if fieldDescr['widgetType'] != 'backField':
user = self.portal_membership.getAuthenticatedMember()
if isEdit:
perm = fieldDescr['atField'].write_permission
else:
perm = fieldDescr['atField'].read_permission
if not user.has_permission(perm, self):
return False
# Evaluate fieldDescr['show']
if callable(fieldDescr['show']):
res = fieldDescr['show'](self.appy())
else:
res = fieldDescr['show']
# Take into account possible values 'view' and 'edit' for 'show' param.
if (res == 'view' and isEdit) or (res == 'edit' and not isEdit):
res = False
return res
def getAppyFields(self, isEdit, page):
'''Returns the fields sorted by group. For every field, a dict
containing the relevant info needed by the view or edit templates is
given.'''
res = []
groups = {} # The already encountered groups
for fieldDescr in self._appy_getOrderedFields(isEdit):
# Select only widgets shown on current page
if fieldDescr.page != page: continue
# Do not take into account hidden fields and fields that can't be
# edited through the edit view
if not self.showField(fieldDescr, isEdit): continue
if not fieldDescr.group:
res.append(fieldDescr.get())
else:
# Have I already met this group?
groupName, cols = GroupDescr.getGroupInfo(fieldDescr.group)
if not groups.has_key(groupName):
groupDescr = GroupDescr(groupName, cols,
fieldDescr.appyType['page']).get()
groups[groupName] = groupDescr
res.append(groupDescr)
else:
groupDescr = groups[groupName]
groupDescr['fields'].append(fieldDescr.get())
if groups:
for groupDict in groups.itervalues():
GroupDescr.computeRows(groupDict)
return res
def getAppyStates(self, phase, currentOnly=False):
'''Returns information about the states that are related to p_phase.
If p_currentOnly is True, we return the current state, even if not
related to p_phase.'''
res = []
dcWorkflow = self.getWorkflow(appy=False)
if not dcWorkflow: return res
currentState = self.portal_workflow.getInfoFor(self, 'review_state')
if currentOnly:
return [StateDescr(currentState,'current').get()]
workflow = self.getWorkflow(appy=True)
if workflow:
stateStatus = 'done'
for stateName in workflow._states:
if stateName == currentState:
stateStatus = 'current'
elif stateStatus != 'done':
stateStatus = 'future'
state = getattr(workflow, stateName)
if (state.phase == phase) and \
(self._appy_showState(workflow, state.show)):
res.append(StateDescr(stateName, stateStatus).get())
return res
def getAppyTransitions(self):
'''Returns the transitions that the user can trigger on p_self.'''
transitions = self.portal_workflow.getTransitionsFor(self)
res = []
if transitions:
# Retrieve the corresponding Appy transition, to check if the user
# may view it.
workflow = self.getWorkflow(appy=True)
if not workflow: return transitions
for transition in transitions:
# Get the corresponding Appy transition
appyTr = workflow._transitionsMapping[transition['id']]
if self._appy_showTransition(workflow, appyTr.show):
res.append(transition)
return res
def getAppyPage(self, isEdit, phaseInfo, appyName=True):
'''On which page am I? p_isEdit indicates if the current page is an
edit or consult view. p_phaseInfo indicates the current phase.'''
pageAttr = 'pageName'
if isEdit:
pageAttr = 'fieldset' # Archetypes page name
default = phaseInfo['pages'][0]
# Default page is the first page of the current phase
res = self.REQUEST.get(pageAttr, default)
if appyName and (res == 'default'):
res = 'main'
return res
def getAppyPages(self, phase='main'):
'''Gets the list of pages that are defined for this content type.'''
res = []
for atField in self.Schema().filterFields(isMetadata=0):
appyType = self.getAppyType(atField.getName())
if not appyType: continue
if (appyType['phase'] == phase) and (appyType['page'] not in res) \
and self._appy_showPage(appyType['page'], appyType['pageShow']):
res.append(appyType['page'])
for appyType, fieldRel in self.getAppyBackRefs():
if (appyType['backd']['phase'] == phase) and \
(appyType['backd']['page'] not in res) and \
self._appy_showPage(appyType['backd']['page'],
appyType['backd']['pageShow']):
res.append(appyType['backd']['page'])
return res
def getAppyPhases(self, currentOnly=False, fieldset=None, forPlone=False):
'''Gets the list of phases that are defined for this content type. If
p_currentOnly is True, the search is limited to the current phase.
If p_fieldset is not None, the search is limited to the phase
corresponding the Plone fieldset whose name is given in this
parameter. If p_forPlone=True, among phase info we write Plone
fieldset names, which are a bit different from Appy page names.'''
# Get the list of phases
res = [] # Ordered list of phases
phases = {} # Dict of phases
for atField in self.Schema().filterFields(isMetadata=0):
appyType = self.getAppyType(atField.getName())
if not appyType: continue
if appyType['phase'] not in phases:
phase = PhaseDescr(appyType['phase'],
self.getAppyStates(appyType['phase']), forPlone, self)
res.append(phase.__dict__)
phases[appyType['phase']] = phase
else:
phase = phases[appyType['phase']]
phase.addPage(appyType, self)
for appyType, fieldRel in self.getAppyBackRefs():
if appyType['backd']['phase'] not in phases:
phase = PhaseDescr(appyType['backd']['phase'],
self.getAppyStates(appyType['backd']['phase']),
forPlone, self)
res.append(phase.__dict__)
phases[appyType['phase']] = phase
else:
phase = phases[appyType['backd']['phase']]
phase.addPage(appyType['backd'], self)
# Remove phases that have no visible page
for i in range(len(res)-1, -1, -1):
if not res[i]['pages']:
del phases[res[i]['name']]
del res[i]
# Then, compute status of phases
for ph in phases.itervalues():
ph.computeStatus()
ph.totalNbOfPhases = len(res)
# Restrict the result if we must not produce the whole list of phases
if currentOnly:
for phaseInfo in res:
if phaseInfo['phaseStatus'] == 'Current':
return phaseInfo
elif fieldset:
for phaseInfo in res:
if fieldset in phaseInfo['pages']:
return phaseInfo
else:
return res
def changeRefOrder(self, fieldName, objectUid, newIndex, isDelta):
'''This method changes the position of object with uid p_objectUid in
reference field p_fieldName to p_newIndex i p_isDelta is False, or
to actualIndex+p_newIndex if p_isDelta is True.'''
sortedFieldName = '_appy_%s' % fieldName
sortedObjectsUids = getattr(self, sortedFieldName)
oldIndex = sortedObjectsUids.index(objectUid)
sortedObjectsUids.remove(objectUid)
if isDelta:
newIndex = oldIndex + newIndex
else:
pass # To implement later on
sortedObjectsUids.insert(newIndex, objectUid)
def onChangeRefOrder(self):
'''This method is called when the user wants to change order of an
item in a reference field.'''
rq = self.REQUEST
# Move the item up (-1), down (+1) ?
move = -1 # Move up
if rq['move'] == 'down':
move = 1 # Down
isDelta = True
self.changeRefOrder(rq['fieldName'], rq['refObjectUid'], move, isDelta)
def getWorkflow(self, appy=True):
'''Returns the Appy workflow instance that is relevant for this
object. If p_appy is False, it returns the DC workflow.'''
res = None
if appy:
# Get the workflow class first
workflowClass = None
if self.wrapperClass:
appyClass = self.wrapperClass.__bases__[1]
if hasattr(appyClass, 'workflow'):
workflowClass = appyClass.workflow
if workflowClass:
# Get the corresponding prototypical workflow instance
res = self.getProductConfig().workflowInstances[workflowClass]
else:
dcWorkflows = self.portal_workflow.getWorkflowsFor(self)
if dcWorkflows:
res = dcWorkflows[0]
return res
def getWorkflowLabel(self, stateName=None):
'''Gets the i18n label for the workflow current state. If no p_stateName
is given, workflow label is given for the current state.'''
res = ''
wf = self.getWorkflow(appy=False)
if wf:
res = stateName
if not res:
res = self.portal_workflow.getInfoFor(self, 'review_state')
appyWf = self.getWorkflow(appy=True)
if appyWf:
res = '%s_%s' % (wf.id, res)
return res
def hasHistory(self):
'''Has this object an history?'''
if hasattr(self.aq_base, 'workflow_history') and self.workflow_history:
key = self.workflow_history.keys()[0]
for event in self.workflow_history[key]:
if event['action'] and (event['comments'] != '_invisible_'):
return True
return False
def getHistory(self, startNumber=0, reverse=True, includeInvisible=False):
'''Returns the history for this object, sorted in reverse order (most
recent change first) if p_reverse is True.'''
batchSize = 5
key = self.workflow_history.keys()[0]
history = list(self.workflow_history[key][1:])
if not includeInvisible:
history = [e for e in history if e['comments'] != '_invisible_']
if reverse: history.reverse()
return {'events': history[startNumber:startNumber+batchSize],
'totalNumber': len(history), 'batchSize':batchSize}
def getComputedValue(self, appyType):
'''Computes on p_self the value of the Computed field corresponding to
p_appyType.'''
res = ''
obj = self._appy_getWrapper(force=True)
if appyType['method']:
try:
res = appyType['method'](obj)
if not isinstance(res, basestring):
res = repr(res)
except Exception, e:
res = str(e)
return res
def may(self, transitionName):
'''May the user execute transition named p_transitionName?'''
# Get the Appy workflow instance
workflow = self.getWorkflow()
res = False
if workflow:
# Get the corresponding Appy transition
transition = workflow._transitionsMapping[transitionName]
user = self.portal_membership.getAuthenticatedMember()
if isinstance(transition.condition, basestring):
# It is a role. Transition may be triggered if the user has this
# role.
res = user.has_role(transition.condition, self)
elif type(transition.condition) == types.FunctionType:
obj = self._appy_getWrapper()
res = transition.condition(workflow, obj)
elif type(transition.condition) in (tuple, list):
# It is a list of roles and or functions. Transition may be
# triggered if user has at least one of those roles and if all
# functions return True.
hasRole = None
for roleOrFunction in transition.condition:
if isinstance(roleOrFunction, basestring):
if hasRole == None:
hasRole = False
if user.has_role(roleOrFunction, self):
hasRole = True
elif type(roleOrFunction) == types.FunctionType:
obj = self._appy_getWrapper()
if not roleOrFunction(workflow, obj):
return False
if hasRole != False:
res = True
return res
def executeAppyAction(self, actionName, reindex=True):
'''Executes action with p_fieldName on this object.'''
appyClass = self.wrapperClass.__bases__[1]
appyType = getattr(appyClass, actionName)
actionRes = appyType(self._appy_getWrapper(force=True))
self.reindexObject()
return appyType.result, actionRes
def onExecuteAppyAction(self):
'''This method is called every time a user wants to execute an Appy
action on an object.'''
rq = self.REQUEST
resultType, actionResult = self.executeAppyAction(rq['fieldName'])
successfull, msg = actionResult
if not msg:
# Use the default i18n messages
suffix = 'ko'
if successfull:
suffix = 'ok'
label='%s_action_%s' % (self.getLabelPrefix(rq['fieldName']),suffix)
msg = self.translate(label)
if (resultType == 'computation') or not successfull:
self.plone_utils.addPortalMessage(msg)
return self.goto(rq['HTTP_REFERER'], True)
else:
# msg does not contain a message, but a complete file to show as is.
# (or, if your prefer, the message must be shown directly to the
# user, not encapsulated in a Plone page).
res = self.getProductConfig().File(msg.name, msg.name, msg,
content_type=mimetypes.guess_type(msg.name)[0])
return res.index_html(rq, rq.RESPONSE)
def onTriggerTransition(self):
'''This method is called whenever a user wants to trigger a workflow
transition on an object.'''
rq = self.REQUEST
self.portal_workflow.doActionFor(self, rq['workflow_action'],
comment = rq.get('comment', ''))
# Where to redirect the user back ?
urlBack = rq['HTTP_REFERER']
if urlBack.find('?') != -1:
# Remove params; this way, the user may be redirected to correct
# phase when relevant.
urlBack = urlBack[:urlBack.find('?')]
msg = self.translate(u'Your content\'s status has been modified.',
domain='plone')
self.plone_utils.addPortalMessage(msg)
self.reindexObject()
return self.goto(urlBack)
def callAppySelect(self, selectMethod, brains):
'''Selects objects from a Reference field.'''
if selectMethod:
obj = self._appy_getWrapper(force=True)
allObjects = [b.getObject()._appy_getWrapper() \
for b in brains]
filteredObjects = selectMethod(obj, allObjects)
filteredUids = [o.o.UID() for o in filteredObjects]
res = []
for b in brains:
if b.UID in filteredUids:
res.append(b)
else:
res = brains
return res
def getCssClasses(self, appyType, asSlave=True):
'''Gets the CSS classes (used for master/slave relationships, or if the
field corresponding to p_appyType is focus) for this object,
either as slave (p_asSlave=True) or as master. The HTML element on
which to define the CSS class for a slave or a master is
different. So this method is called either for getting CSS classes
as slave or as master. We set the focus-specific CSS class only when
p_asSlave is True, because we this place as being the "standard" one
for specifying CSS classes for a field.'''
res = ''
if not asSlave and appyType['slaves']:
res = 'appyMaster master_%s' % appyType['id']
elif asSlave and appyType['master']:
res = 'slave_%s' % appyType['master'].id
res += ' slaveValue_%s_%s' % (appyType['master'].id,
appyType['masterValue'])
# Add the focus-specific class if needed
if appyType['focus']:
prefix = ''
if res: prefix = ' '
res += prefix + 'appyFocus'
return res
def fieldValueSelected(self, fieldName, value, vocabValue):
'''When displaying a selection box (ie a String with a validator being a
list), must the _vocabValue appear as selected?'''
# Check according to database value
if (type(value) in sequenceTypes):
if vocabValue in value: return True
else:
if vocabValue == value: return True
# Check according to value in request
valueInReq = self.REQUEST.get(fieldName, None)
if type(valueInReq) in sequenceTypes:
if vocabValue in valueInReq: return True
else:
if vocabValue == valueInReq: return True
return False
def checkboxChecked(self, fieldName, value):
'''When displaying a checkbox, must it be checked or not?'''
valueInReq = self.REQUEST.get(fieldName, None)
if valueInReq != None:
return valueInReq in ('True', 1, '1')
else:
return value
def getLabelPrefix(self, fieldName=None):
'''For some i18n labels, wee need to determine a prefix, which may be
linked to p_fieldName. Indeed, the prefix may be based on the name
of the (super-)class where p_fieldName is defined.'''
res = self.meta_type
if fieldName:
appyType = self.getAppyType(fieldName)
res = '%s_%s' % (self._appy_getAtType(appyType['selfClass']),
fieldName)
return res
def _appy_getWrapper(self, force=False):
'''Returns the wrapper object for p_self. It is created if it did not
exist.'''
if (not hasattr(self.aq_base, 'appyWrapper')) or force:
# In some cases (p_force=True), we need to re-generate the
# wrapper object. Else, acquisition may be lost on wrapper.o.
self.appyWrapper = self.wrapperClass(self)
return self.appyWrapper
def appy(self):
'''Nice alias to the previous method.'''
return self._appy_getWrapper(force=True)
def _appy_getSourceClass(self, fieldName, baseClass):
'''We know that p_fieldName was defined on Python class p_baseClass or
one of its parents. This method returns the exact class (p_baseClass
or a parent) where it was defined.'''
if fieldName in baseClass.__dict__:
return baseClass
else:
return self._appy_getSourceClass(fieldName, baseClass.__bases__[0])
def _appy_getTypeAsDict(self, fieldName, appyType, baseClass):
'''Within page templates, the appyType is given as a dict instead of
an object in order to avoid security problems.'''
appyType.selfClass = self._appy_getSourceClass(fieldName, baseClass)
res = appyType.__dict__
if res.has_key('back') and res['back'] and (not res.has_key('backd')):
res['backd'] = res['back'].__dict__
# I create a new entry "backd"; if I put the dict in "back" I
# really modify the initial appyType object and I don't want to do
# this.
# Add the i18n label for the field
if not res.has_key('label'):
res['label'] = '%s_%s' % (self._appy_getAtType(appyType.selfClass),
fieldName)
return res
def _appy_getAtType(self, appyClass, flavour=None):
'''Gets the name of the Archetypes class that corresponds to
p_appyClass (which is a Python class coming from the user
application). If p_flavour is specified, the method returns the name
of the specific Archetypes class in this flavour (ie suffixed with
the flavour number).'''
res = ArchetypesClassDescriptor.getClassName(appyClass)
appName = self.getProductConfig().PROJECTNAME
if res.find('Extensions_appyWrappers') != -1:
# This is not a content type defined Maybe I am a tool or flavour
res = appName + appyClass.__name__
elif issubclass(appyClass, appy.gen.Tool):
# This is the custom tool
res = '%sTool' % appName
elif issubclass(appyClass, appy.gen.Flavour):
# This is the custom Flavour
res = '%sFlavour' % appName
else:
if flavour and flavour.number != 1:
res += '_%d' % flavour.number
return res
def _appy_getRefsBack(self, fieldName, relName, ploneObjects=False,
noListIfSingleObj=False):
'''This method returns the list of objects linked to this one
through the BackRef corresponding to the Archetypes
relationship named p_relName.'''
res = []
referers = self.getProductConfig().referers
objs = self.getBRefs(relName)
for obj in objs:
if not ploneObjects:
obj = obj._appy_getWrapper(force=True)
res.append(obj)
if res and noListIfSingleObj:
className = self.__class__.__name__
appyType = None
for anAppyType, rel in referers[className]:
if rel == relName:
appyType = anAppyType
break
if appyType.back.multiplicity[1] == 1:
res = res[0]
return res
def _appy_showPage(self, page, pageShow):
'''Must I show p_page?'''
if callable(pageShow):
return pageShow(self._appy_getWrapper(force=True))
else: return pageShow
def _appy_showState(self, workflow, stateShow):
'''Must I show a state whose "show value" is p_stateShow?'''
if callable(stateShow):
return stateShow(workflow, self._appy_getWrapper())
else: return stateShow
def _appy_showTransition(self, workflow, transitionShow):
'''Must I show a transition whose "show value" is p_transitionShow?'''
if callable(transitionShow):
return transitionShow(workflow, self._appy_getWrapper())
else: return transitionShow
def _appy_managePermissions(self):
'''When an object is created or updated, we must update "add"
permissions accordingly: if the object is a folder, we must set on
it permissions that will allow to create, inside it, objects through
Ref fields; if it is not a folder, we must update permissions on its
parent folder instead.'''
# Determine on which folder we need to set "add" permissions
folder = self
if not self.isPrincipiaFolderish:
folder = self.getParentNode()
# On this folder, set "add" permissions for every content type that will
# be created through reference fields
allCreators = set()
for field in self.schema.fields():
if field.type == 'reference':
refContentTypeName= self.getAppyRefPortalType(field.getName())
refContentType = getattr(self.portal_types, refContentTypeName)
refMetaType = refContentType.content_meta_type
if refMetaType in self.getProductConfig(\
).ADD_CONTENT_PERMISSIONS:
# No specific "add" permission is defined for tool and
# flavour, for example.
appyClass = refContentType.wrapperClass.__bases__[-1]
# Get roles that may add this content type
creators = getattr(appyClass, 'creators', None)
if not creators:
creators = self.getProductConfig().defaultAddRoles
allCreators = allCreators.union(creators)
# Grant this "add" permission to those roles
updateRolesForPermission(
self.getProductConfig().ADD_CONTENT_PERMISSIONS[\
refMetaType], creators, folder)
# Beyond content-type-specific "add" permissions, creators must also
# have the main permission "Add portal content".
if allCreators:
updateRolesForPermission('Add portal content', tuple(allCreators),
folder)
def _appy_getDisplayList(self, values, labels, domain):
'''Creates a DisplayList given a list of p_values and corresponding
i18n p_labels.'''
res = []
i = -1
for v in values:
i += 1
res.append( (v, self.utranslate(labels[i], domain=domain)))
return self.getProductConfig().DisplayList(tuple(res))
def _appy_getDynamicDisplayList(self, methodName):
'''Calls the method named p_methodName for producing a DisplayList from
values computed dynamically. If methodName begins with _appy_, it is
a special Appy method: we will call it on the Mixin directly. Else,
it is a user method: we will call it on the wrapper. Some args can
be hidden into p_methodName, separated with stars, like in this
example: method1*arg1*arg2. Only string params are supported.'''
# Unwrap parameters if any.
if methodName.find('*') != -1:
elems = methodName.split('*')
methodName = elems[0]
args = elems[1:]
else:
args = ()
# On what object must be call the method that will produce the values?
obj = self
if methodName.startswith('tool:'):
obj = self.getTool()
methodName = methodName[5:]
# Do we need to call the method on the object or on the wrapper?
if methodName.startswith('_appy_'):
exec 'res = obj.%s(*args)' % methodName
else:
exec 'res = obj.appy().%s(*args)' % methodName
return self.getProductConfig().DisplayList(tuple(res))
nullValues = (None, '', ' ')
numbersMap = {'Integer': 'int', 'Float': 'float'}
validatorTypes = (types.FunctionType, type(String.EMAIL))
def _appy_validateField(self, fieldName, value, label, specificType):
'''Checks whether the p_value entered in field p_fieldName is
correct.'''
appyType = self.getAppyType(fieldName)
msgId = None
if (specificType == 'Ref') and appyType['link']:
# We only check "link" Refs because in edit views, "add" Refs are
# not visible. So if we check "add" Refs, on an "edit" view we will
# believe that that there is no referred object even if there is.
# If the field is a reference, appy must ensure itself that
# multiplicities are enforced.
fieldValue = self.REQUEST.get('appy_ref_%s' % fieldName, '')
if not fieldValue:
nbOfRefs = 0
elif isinstance(fieldValue, basestring):
nbOfRefs = 1
else:
nbOfRefs = len(fieldValue)
minRef = appyType['multiplicity'][0]
maxRef = appyType['multiplicity'][1]
if maxRef == None:
maxRef = sys.maxint
if nbOfRefs < minRef:
msgId = 'min_ref_violated'
elif nbOfRefs > maxRef:
msgId = 'max_ref_violated'
elif specificType in self.numbersMap: # Float, Integer
pyType = self.numbersMap[specificType]
# Validate only if input value is there.
# By the way, we also convert the value.
if value not in self.nullValues:
try:
exec 'value = %s(value)' % pyType
except ValueError:
msgId = 'bad_%s' % pyType
else:
value = None
# Apply the custom validator if it exists
validator = appyType['validator']
if not msgId and (type(validator) in self.validatorTypes):
obj = self.appy()
if type(validator) == self.validatorTypes[0]:
# It is a custom function. Execute it.
try:
validValue = validator(obj, value)
if isinstance(validValue, basestring) and validValue:
# Validation failed; and p_validValue contains an error
# message.
return validValue
else:
if not validValue:
msgId = label
except Exception, e:
return str(e)
except:
msgId = label
elif type(validator) == self.validatorTypes[1]:
# It is a regular expression
if (value not in self.nullValues) and \
not validator.match(value):
# If the regular expression is among the default ones, we
# generate a specific error message.
if validator == String.EMAIL:
msgId = 'bad_email'
elif validator == String.URL:
msgId = 'bad_url'
elif validator == String.ALPHANUMERIC:
msgId = 'bad_alphanumeric'
else:
msgId = label
res = msgId
if msgId:
res = self.utranslate(msgId, domain=self.i18nDomain)
return res
def validateAllFields(self, REQUEST, errors):
'''This method is called when individual validation of all fields
succeed (when editing or creating an object). Then, this method
performs inter-field validation. This way, the user must first
correct individual fields before being confronted to potential
inter-fields validation errors.'''
obj = self.appy()
if not hasattr(obj, 'validate'): return
appyRequest = getAppyRequest(REQUEST, obj)
appyErrors = ValidationErrors()
obj.validate(appyRequest, appyErrors)
# This custom "validate" method may have added fields in the given
# ValidationErrors instance. Now we must fill the Zope "errors" dict
# based on it. For every error message that is not a string,
# we replace it with the standard validation error for the
# corresponding field.
for key, value in appyErrors.__dict__.iteritems():
resValue = value
if not isinstance(resValue, basestring):
msgId = '%s_valid' % self.getLabelPrefix(key)
resValue = self.utranslate(msgId, domain=self.i18nDomain)
errors[key] = resValue
def _appy_getPortalType(self, request):
'''Guess the portal_type of p_self from info about p_self and
p_request.'''
res = None
# If the object is being created, self.portal_type is not correctly
# initialized yet.
if request.has_key('__factory__info__'):
factoryInfo = request['__factory__info__']
if factoryInfo.has_key('stack'):
res = factoryInfo['stack'][0]
if not res:
res = self.portal_type
return res
def _appy_generateDocument(self):
'''Generates the document from a template whose UID is specified in the
request for a given object whose UID is also in the request.'''
# Get the object
objectUid = self.REQUEST.get('objectUid')
obj = self.uid_catalog(UID=objectUid)[0].getObject()
# Get the POD template
templateUid = self.REQUEST.get('templateUid')
podTemplate = self.uid_catalog(UID=templateUid)[0].getObject()
return podTemplate.generateDocument(obj)
def _appy_manageSortedRefs(self):
'''For every reference field, this method creates the additional
reference lists that are ordered (if it did not already exist).'''
for field in self.schema.fields():
if field.type == 'reference':
sortedRefField = '_appy_%s' % field.getName()
if not hasattr(self.aq_base, sortedRefField):
pList = self.getProductConfig().PersistentList
exec 'self.%s = pList()' % sortedRefField
def _appy_manageRefs(self, created):
'''Every time an object is created or updated, this method updates
the Reference fields accordingly.'''
self._appy_manageSortedRefs()
self._appy_manageRefsFromRequest()
# If the creation was initiated by another object, update the
# reference.
if created and hasattr(self.REQUEST, 'SESSION'):
# When used by the test system, no SESSION object is created.
session = self.REQUEST.SESSION
initiatorUid = session.get('initiator', None)
initiator = None
if initiatorUid:
initiatorRes = self.uid_catalog.searchResults(UID=initiatorUid)
if initiatorRes:
initiator = initiatorRes[0].getObject()
if initiator:
fieldName = session.get('initiatorField')
initiator._appy_getWrapper(force=True).link(fieldName, self)
# Re-initialise the session
session['initiator'] = None
def _appy_manageRefsFromRequest(self):
'''Appy manages itself some Ref fields (with link=True). So here we must
update the Ref fields.'''
fieldsInRequest = [] # Fields present in the request
for requestKey in self.REQUEST.keys():
if requestKey.startswith('appy_ref_'):
fieldName = requestKey[9:]
fieldsInRequest.append(fieldName)
fieldValue = self.REQUEST[requestKey]
sortedRefField = getattr(self, '_appy_%s' % fieldName)
del sortedRefField[:]
if not fieldValue: fieldValue = []
if isinstance(fieldValue, basestring):
fieldValue = [fieldValue]
refObjects = []
for uid in fieldValue:
obj = self.uid_catalog(UID=uid)[0].getObject()
refObjects.append(obj)
sortedRefField.append(uid)
exec 'self.set%s%s(refObjects)' % (fieldName[0].upper(),
fieldName[1:])
# Manage Ref fields that are not present in the request
currentFieldset = self.REQUEST.get('fieldset', 'default')
for field in self.schema.fields():
if (field.type == 'reference') and \
(field.schemata == currentFieldset) and \
(field.getName() not in fieldsInRequest):
# If this field is visible, it was not present in the request:
# it means that we must remove any Ref from it.
fieldName = field.getName()
appyType = self.getAppyType(fieldName)
fieldDescr = FieldDescr(field, appyType, None)
if self.showField(fieldDescr, isEdit=True):
exec 'self.set%s%s([])' % (fieldName[0].upper(),
fieldName[1:])
def getUrl(self, t='view', **kwargs):
'''This method returns various URLs about this object.'''
baseUrl = self.absolute_url()
params = ''
rq = self.REQUEST
for k, v in kwargs.iteritems(): params += '&%s=%s' % (k, v)
if params: params = params[1:]
if t == 'showRef':
chunk = '/skyn/ajax?objectUid=%s&page=ref&' \
'macro=showReferenceContent&' % self.UID()
startKey = '%s%s_startNumber' % (self.UID(), kwargs['fieldName'])
if rq.has_key(startKey) and not kwargs.has_key(startKey):
params += '&%s=%s' % (startKey, rq[startKey])
return baseUrl + chunk + params
elif t == 'showHistory':
chunk = '/skyn/ajax?objectUid=%s&page=macros&macro=history' % \
self.UID()
if params: params = '&' + params
return baseUrl + chunk + params
else: # We consider t=='view'
return baseUrl + '/skyn/view' + params
def translate(self, label, mapping={}, domain=None, default=None):
'''Translates a given p_label into p_domain with p_mapping.'''
cfg = self.getProductConfig()
if not domain: domain = cfg.PROJECTNAME
return self.translation_service.utranslate(
domain, label, mapping, self, default=default)
# ------------------------------------------------------------------------------