[Zope3-checkins] CVS: Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub - LocalHubEvent.py:1.1 LocalObjectHub.py:1.1 __init__.py:1.1 configure.zcml:1.1 objecthub.gif:1.1
Gary Poster
gary@modernsongs.com
Mon, 21 Oct 2002 02:14:47 -0400
Update of /cvs-repository/Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub
In directory cvs.zope.org:/tmp/cvs-serv13005/Zope/App/OFS/Services/LocalObjectHub
Added Files:
LocalHubEvent.py LocalObjectHub.py __init__.py configure.zcml
objecthub.gif
Log Message:
sorry for the huge honking checkin.
Adds a simple local objecthub implementation and made ObjectHub a service
Modifies the main objecthub as we have been discussing:
* objecthub attribute is hubid, not hid (sorry Jim, I'll change it back if you want but there were a lot of "yay"s and no "nay"s :-)
* no more IObjectAddedHubEvent
* IObjectRemovedEvent now (hopefully) actually has the effect on the ObjectHub that is described in the interface, and that we agreed upon, namely (assuming removed object was cataloged in objecthub) removing catalog of object in objecthub and sending out an IObjectRemovedHubEvent, subclass of IObjectUnregisteredHubEvent, to the ObjectHub subscribers
I tried to spruce up the LocalEventService a bit but the code still looks as opaque as ever, I'm afraid. Among other small improvements, though, at least you actually can see the silly "user interface" now without a traceback. Now for a *real* user interface sometime. :-)
Fixed a few typos while I was at it as well...and I'm sure made my share of new ones :-)
=== Added File Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/LocalHubEvent.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""
Revision information:
$Id: LocalHubEvent.py,v 1.1 2002/10/21 06:14:46 poster Exp $
"""
__metaclass__ = type
from Zope.ObjectHub.IHubEvent import IObjectRegisteredHubEvent
from Zope.ObjectHub.IHubEvent import IObjectUnregisteredHubEvent
from Zope.ObjectHub.IHubEvent import IObjectModifiedHubEvent
from Zope.ObjectHub.IHubEvent import IObjectMovedHubEvent
from Zope.ObjectHub.IHubEvent import IObjectRemovedHubEvent
from Zope.App.Traversing import traverse
class HubEvent:
"""Convenient mix-in for HubEvents"""
location = None
hubid = None
def __init__(self, objecthub, hubid, location):
# we keep all three, to avoid unnecessary lookups
# and to give the objecthub an opportunity to do
# caching of objects
self.__objecthub = objecthub
self.hubid = hubid
self.location = location
def __getObject(self):
if hasattr(self, '_v_object'):
return self._v_object
obj = self._v_object = traverse(
self.__objecthub, self.location)
# we use the above instead of the below primarily because
# the object hub call is not guaranteed to work on an
# unregistered event; the above also does a bit less work:
# obj = self._v_object = (self.__objecthub.getObject(self.__hubid)
# and that, unfortunately, is the only reason why we're not
# using the Zope.ObjectHub.HubEvent versions of these...
return obj
object = property(__getObject)
class ObjectRegisteredHubEvent(HubEvent):
"""An ruid has been freshly created and mapped against an object."""
__implements__ = IObjectRegisteredHubEvent
class ObjectUnregisteredHubEvent(HubEvent):
"""We are no longer interested in this object."""
__implements__ = IObjectUnregisteredHubEvent
class ObjectModifiedHubEvent(HubEvent):
"""An object with an ruid has been modified."""
__implements__ = IObjectModifiedHubEvent
class ObjectMovedHubEvent(HubEvent):
"""An object with an ruid has had its context changed. Typically, this
means that it has been moved."""
__implements__ = IObjectMovedHubEvent
class ObjectRemovedHubEvent:
"""An object with an ruid has been removed."""
__implements__ = IObjectRemovedHubEvent
# ...which is a subclass of IObjectUnregisteredHubEvent
def __init__(self, obj, hubid, location):
self.object = obj
self.hubid = hubid
self.location = location
=== Added File Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/LocalObjectHub.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""
Revision information:
$Id: LocalObjectHub.py,v 1.1 2002/10/21 06:14:46 poster Exp $
"""
from Zope.App.OFS.Services.LocalEventService.LocalServiceSubscribable \
import LocalServiceSubscribable
from Zope.App.OFS.Services.LocalEventService.ProtoServiceEventChannel \
import ProtoServiceEventChannel
from Zope.ObjectHub.ObjectHub import ObjectHubError, randid
from Zope.ObjectHub.IObjectHub import IObjectHub
from LocalHubEvent import ObjectRegisteredHubEvent
from LocalHubEvent import ObjectUnregisteredHubEvent
from LocalHubEvent import ObjectModifiedHubEvent
from LocalHubEvent import ObjectMovedHubEvent
from LocalHubEvent import ObjectRemovedHubEvent
from Zope.ObjectHub.IHubEvent import IHubEvent
from Zope.Exceptions import NotFoundError
from Zope.Event.IObjectEvent import \
IObjectEvent, IObjectAddedEvent, IObjectModifiedEvent
from Zope.Event.IObjectEvent import IObjectRemovedEvent, IObjectMovedEvent
from Persistence.BTrees.IOBTree import IOBTree
from Persistence.BTrees.OIBTree import OIBTree
from Zope.ContextWrapper import ContextMethod
from Zope.Proxy.ContextWrapper import isWrapper
from Zope.App.Traversing import getPhysicalPathString
from Zope.App.Traversing import locationAsUnicode
from Zope.Proxy.ProxyIntrospection import removeAllProxies
from Zope.Proxy.ContextWrapper import ContextWrapper
class ILocalObjectHub(IObjectHub): # XXX also put in proto stuff here?
def lookupHubId(wrappedObj_or_location):
"""like IObjectHub.lookupHubId but also accepts wrapped object"""
def register(wrappedObj_or_location):
"""like IObjectHub.register but also accepts wrapped object"""
def unregister(wrappedObj_or_location_or_hubid):
"""like IObjectHub.unregister but also accepts wrapped object"""
class LocalObjectHub(ProtoServiceEventChannel):
# this implementation makes the decision to not interact with any
# object hubs above it: it is a world unto itself, as far as it is
# concerned, and if it doesn't know how to do something, it won't
# ask anything else to try. Everything else is YAGNI for now.
__implements__ = (
ILocalObjectHub,
ProtoServiceEventChannel.__implements__)
def __init__(self):
ProtoServiceEventChannel.__init__(self)
self.__hubid_to_location = IOBTree()
self.__location_to_hubid = OIBTree()
def _notify(clean_self, wrapped_self, event):
subscriptionses = clean_self.subscriptionsForEvent(event)
# that's a non-interface shortcut for
# subscriptionses = clean_self._registry.getAllForObject(event)
for subscriptions in subscriptionses:
for subscriber,filter in subscriptions:
if filter is not None and not filter(event):
continue
ContextWrapper(subscriber, wrapped_self).notify(event)
# notify has to have a minor overhaul from the placeless version
def notify(wrapped_self, event):
'''See interface ISubscriber'''
clean_self = removeAllProxies(wrapped_self)
clean_self._notify(wrapped_self, event)
if IObjectEvent.isImplementedBy(event):
# generate NotificationHubEvents only if object is known
# ie registered
if IObjectMovedEvent.isImplementedBy(event):
canonical_location = locationAsUnicode(event.fromLocation)
hubid = clean_self._lookupHubId(canonical_location)
if hubid is not None:
canonical_new_location = locationAsUnicode(
event.location)
location_to_hubid = clean_self.__location_to_hubid
if location_to_hubid.has_key(canonical_new_location):
raise ObjectHubError(
'Cannot move to location %s, '
'as there is already something there'
% canonical_new_location)
hubid = location_to_hubid[canonical_location]
del location_to_hubid[canonical_location]
location_to_hubid[canonical_new_location] = hubid
clean_self.__hubid_to_location[hubid] = (
canonical_new_location)
# send out IObjectMovedHubEvent to plugins
event = ObjectMovedHubEvent(
wrapped_self,
hubid,
canonical_new_location)
clean_self._notify(wrapped_self, event)
else:
canonical_location = locationAsUnicode(event.location)
hubid = clean_self._lookupHubId(canonical_location)
if hubid is not None:
if IObjectModifiedEvent.isImplementedBy(event):
# send out IObjectModifiedHubEvent to plugins
event = ObjectModifiedHubEvent(
wrapped_self,
hubid,
canonical_location)
clean_self._notify(wrapped_self, event)
elif IObjectRemovedEvent.isImplementedBy(event):
del clean_self.__hubid_to_location[hubid]
del clean_self.__location_to_hubid[canonical_location]
# send out IObjectRemovedHubEvent to plugins
event = ObjectRemovedHubEvent(
event.object,
hubid,
canonical_location)
clean_self._notify(wrapped_self, event)
notify = ContextMethod(notify)
# lookupHubId just has new ability to take an object
def lookupHubId(self, location):
'''See interface ILocalObjectHub'''
if isWrapper(location):
location = getPhysicalPathString(location)
hubid = self._lookupHubId(location)
if hubid is None:
raise NotFoundError, locationAsUnicode(location)
else:
return hubid
def lookupLocation(self, hubid):
'''See interface IObjectHub'''
try:
return self.__hubid_to_location[hubid]
except KeyError:
raise NotFoundError, hubid
def getObject(self, hubid):
'''See interface IObjectHub'''
location = self.lookupLocation(hubid)
adapter = getAdapter(self, ITraverser)
return adapter.traverse(location)
getObject = ContextMethod(getObject)
# we must give register an overhaul also
def register(wrapped_self, location):
'''See interface ILocalObjectHub'''
clean_self = removeAllProxies(wrapped_self)
if isWrapper(location):
location = getPhysicalPathString(location)
canonical_location=locationAsUnicode(location)
if location[0] != u'/':
raise ValueError, "Location must be absolute"
location_to_hubid = clean_self.__location_to_hubid
if location_to_hubid.has_key(canonical_location):
raise ObjectHubError, 'location %s already in object hub' % \
canonical_location
hubid = clean_self._generateHubId(canonical_location)
location_to_hubid[canonical_location] = hubid
# send out IObjectRegisteredHubEvent to plugins
event = ObjectRegisteredHubEvent(
wrapped_self,
hubid,
canonical_location)
clean_self._notify(wrapped_self, event)
return hubid
register = ContextMethod(register)
# as well as unregister
def unregister(wrapped_self, location):
'''See interface ILocalObjectHub'''
clean_self = removeAllProxies(wrapped_self)
if isWrapper(location):
location = getPhysicalPathString(location)
elif isinstance(location, int):
canonical_location=clean_self.lookupLocation(location)
else:
canonical_location=locationAsUnicode(location)
location_to_hubid = clean_self.__location_to_hubid
hubid_to_location = clean_self.__hubid_to_location
try:
hubid = location_to_hubid[canonical_location]
except KeyError:
raise NotFoundError, 'location %s is not in object hub' % \
canonical_location
else:
del hubid_to_location[hubid]
del location_to_hubid[canonical_location]
# send out IObjectUnregisteredHubEvent to plugins
event = ObjectUnregisteredHubEvent(
wrapped_self,
hubid,
canonical_location)
clean_self._notify(wrapped_self, event)
unregister = ContextMethod(unregister)
############################################################
# we use two helpers copied from the ObjectHub base class:
def _generateHubId(self, location):
index=getattr(self, '_v_nextid', 0)
if index%4000 == 0: index = randid()
hubid_to_location=self.__hubid_to_location
while not hubid_to_location.insert(index, location):
index=randid()
self._v_nextid=index+1
return index
def _lookupHubId(self, location):
canonical_location = locationAsUnicode(location)
return self.__location_to_hubid.get(canonical_location, None)
# Not sure about plugins yet--will see what the response to my add
# and remove emails are
=== Added File Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/__init__.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Local Object Hub"""
from Zope.ComponentArchitecture import getService
from Zope.Proxy.ContextWrapper import isWrapper
from Zope.App.Traversing import getPhysicalPathString
from Zope.App.Traversing import locationAsUnicode
def normalizeToHubIds(context, *args):
"""given a context and any number of hub ids, physical paths,
or wrapped objects, returns a normalized list of each item as hubid
using the ObjectHub closest to the context.
"""
obHub = getService(context, "ObjectHub")
args = list(args)
for ix in len(args):
arg = args[ix]
if isinstance(arg, int):
pass
elif isinstance(arg, str):
args[ix] = obHub.lookupHubId(locationAsUnicode(arg))
elif isWrapper(arg):
args[ix] = getPhysicalPathString(arg)
return args
=== Added File Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/configure.zcml ===
<zopeConfigure
xmlns='http://namespaces.zope.org/zope'
xmlns:browser='http://namespaces.zope.org/browser'
xmlns:service='http://namespaces.zope.org/service'
>
<content class='.LocalObjectHub.'>
<factory
id='ObjectHub'
permission='Zope.ManageServices' />
<require
permission="Zope.View"
attributes="notify lookupRuid lookupLocation getObject register unregister" />
<require
permission="Zope.ManageServices"
attributes="bound unbound subscribe unsubscribe subscribeOnBind
unsubscribedFrom subscribedTo" />
</content>
<browser:menuItem menu="add_component" for="Zope.App.OFS.Container.IAdding."
action="ObjectHub" title='ObjectHub'
description='An object hub, for cataloging, unique object ids, and more: use sparingly' />
<browser:icon name="zmi_icon" for="Zope.ObjectHub.IObjectHub."
file="./objecthub.gif" />
<include package=".Views" />
</zopeConfigure>
=== Added File Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/objecthub.gif ===
<Binary-ish file>