[Zope3-checkins] CVS: Zope3/src/zope/app/utility - fixup.py:1.1
utility.py:1.1 __init__.py:1.5 tests.py:1.6
Jim Fulton
jim at zope.com
Thu Apr 15 11:30:03 EDT 2004
Update of /cvs-repository/Zope3/src/zope/app/utility
In directory cvs.zope.org:/tmp/cvs-serv476/src/zope/app/utility
Modified Files:
__init__.py tests.py
Added Files:
fixup.py utility.py
Log Message:
Reimplemented local utility services, based on the local adapter
service and adapter registries.
=== Added File Zope3/src/zope/app/utility/fixup.py ===
##############################################################################
#
# Copyright (c) 2004 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.
#
##############################################################################
"""Fixup old utility services
$Id: fixup.py,v 1.1 2004/04/15 15:29:31 jim Exp $
"""
from zope.app import zapi
from zope.app.site.interfaces import IPossibleSite
from zope.app.utility import LocalUtilityService
from zope.app.annotation.interfaces import IAnnotations
from persistent.dict import PersistentDict
from zope.interface.adapter import Null
from zope.app.registration.registration import NotifyingRegistrationStack
from zope.app.container.interfaces import IContainer
import zope.app.event.function
key = 'zope.app.utility.fixup20040415'
class ImplementorRegistry(object):
"""Temporary replacement of old interface.implementor class
"""
def notify(event):
db = event.database
conn = db.open()
try:
root = conn.root().get('Application')
if root is None:
return # new db
annotations = IAnnotations(root)
if key in annotations:
return
import sys
sys.modules['zope.interface.implementor'] = sys.modules[__name__]
LocalUtilityService.base = zapi.getService(None, 'Utilities')
fixup(root)
del sys.modules['zope.interface.implementor']
del LocalUtilityService.base
annotations[key] = True
finally:
conn.close()
notify = zope.app.event.function.Subscriber(notify)
def fixup(folder):
if IPossibleSite.providedBy(folder):
sm = folder.getSiteManager()
if sm is not None:
for smfolder in sm.values():
for item in smfolder.values():
if isinstance(item, LocalUtilityService):
if hasattr(item, '_utilities'):
fixup_utility_service(sm, item)
for item in folder.values():
if IContainer.providedBy(item):
fixup(item)
def fixup_utility_service(sm, us):
stacks = PersistentDict()
for name, implementor_registry in us._utilities.iteritems():
for (provided, (registered_provided, stack)
) in implementor_registry._reg.iteritems():
if provided is not registered_provided:
continue
newstack = NotifyingRegistrationStack(us)
newstack.data = stack.data
stacks[(False, (), name, provided)] = newstack
del us._utilities
us.__init__(zapi.getService(None, 'Utilities'))
us.stacks = PersistentDict({Null: stacks})
if sm.queryLocalService('Utilities') is us:
us.notifyActivated()
get_transaction().commit()
=== Added File Zope3/src/zope/app/utility/utility.py ===
##############################################################################
# Copyright (c) 2003 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 utility service implementation.
Besides being functional, this module also serves as an example of
creating a local service; see README.txt.
$Id: utility.py,v 1.1 2004/04/15 15:29:31 jim Exp $
"""
from zope.app.adapter.adapter import LocalAdapterService
from zope.app import zapi
from zope.app.registration.registration import ComponentRegistration
from zope.app.utility.interfaces import ILocalUtilityService
from zope.app.utility.interfaces import IUtilityRegistration
from zope.component.utility import UtilityService
import zope.app.site.interfaces
import zope.interface
import zope.interface.adapter
class LocalUtilityService(UtilityService, LocalAdapterService):
"""Local Utility Service
"""
serviceType = zapi.servicenames.Utilities
zope.interface.implementsOnly(
ILocalUtilityService,
zope.app.site.interfaces.ISimpleService,
zope.app.site.interfaces.IBindingAware,
)
def queryRegistrations(self, name, interface):
return self.queryRegistrationsFor(
UtilityRegistration(name, interface, None)
)
def getLocalUtilitiesFor(self, interface):
# This method is deprecated and is temporarily provided for
# backward compatability
from zope.app import zapi
from zope.app.component.nextservice import getNextService
next = getNextService(self, zapi.servicenames.Utilities)
next_utils = dict(next.getUtilitiesFor(interface))
for name, util in self.getUtilitiesFor(interface):
if next_utils.get(name) != util:
yield name, util
def getRegisteredMatching(self, interface=None, name=None):
# doomed lame depreceated method backward-compatibly lame
L = []
for stacks in self.stacks.itervalues():
for ((s, w, reg_name, iface), stack) in stacks.iteritems():
if not stack:
continue
if interface and not iface is interface:
continue
if name is not None and reg_name.find(name) < 0:
continue
L.append((iface, reg_name, stack))
return L
class UtilityRegistration(ComponentRegistration):
"""Utility component registration for persistent components
This registration configures persistent components in packages to
be utilities.
"""
serviceType = zapi.servicenames.Utilities
############################################################
# To make adapter code happy. Are we going too far?
#
required = zope.interface.adapter.Null
with = ()
provided = property(lambda self: self.interface)
factory = property(lambda self: self.getComponent())
#
############################################################
zope.interface.implements(IUtilityRegistration)
component = property(lambda self: self.getComponent())
def __init__(self, name, interface, component_path, permission=None):
ComponentRegistration.__init__(self, component_path, permission)
self.name = name
self.interface = interface
def usageSummary(self):
# Override IRegistration.usageSummary()
component = self.getComponent()
s = self.getInterface().getName()
if self.name:
s += " registered as '%s'" % self.name
s += ", implemented by %s" %component.__class__.__name__
s += " '%s'"%self.componentPath
return s
def getInterface(self):
# ComponentRegistration calls this when you specify a
# permission; it needs the interface to create a security
# proxy for the interface with the given permission.
# XXX Smells like a dead chicken to me.
return self.interface
=== Zope3/src/zope/app/utility/__init__.py 1.4 => 1.5 ===
--- Zope3/src/zope/app/utility/__init__.py:1.4 Sat Mar 13 18:55:29 2004
+++ Zope3/src/zope/app/utility/__init__.py Thu Apr 15 11:29:31 2004
@@ -11,163 +11,7 @@
##############################################################################
"""Local utility service implementation.
-Besides being functional, this module also serves as an example of
-creating a local service; see README.txt.
-
$Id$
"""
-from zope.interface import implements
-from persistent.dict import PersistentDict
-from persistent import Persistent
-from zope.app.component.nextservice import getNextService
-from zope.app.registration.interfaces import IRegistry
-from zope.app.site.interfaces import ISimpleService
-from zope.app.utility.interfaces import \
- IUtilityRegistration, ILocalUtilityService
-from zope.app.servicenames import Utilities
-from zope.app.registration.registration import \
- RegistrationStack, ComponentRegistration
-from zope.component.exceptions import ComponentLookupError
-from zope.interface.implementor import ImplementorRegistry
-from zope.proxy import removeAllProxies
-from zope.app.container.contained import Contained
-
-class LocalUtilityService(Persistent, Contained):
-
- implements(ILocalUtilityService, IRegistry, ISimpleService)
-
- def __init__(self):
- self._utilities = PersistentDict()
-
- def getUtility(self, interface, name=''):
- """See zope.component.interfaces.IUtilityService"""
- utility = self.queryUtility(interface, name=name)
- if utility is None:
- raise ComponentLookupError("utility", interface, name)
- return utility
-
- def queryUtility(self, interface, default=None, name=''):
- """See zope.component.interfaces.IUtilityService"""
- stack = self.queryRegistrations(name, interface)
- if stack is not None:
- registration = stack.active()
- if registration is not None:
- return registration.getComponent()
-
- next = getNextService(self, Utilities)
- return next.queryUtility(interface, default, name)
-
- def getUtilitiesFor(self, interface):
- """See zope.component.interfaces.IUtilityService"""
- utilities = self.getLocalUtilitiesFor(interface)
- names = map(lambda u: u[0], utilities)
-
- next = getNextService(self, Utilities)
- for utility in next.getUtilitiesFor(interface):
- if utility[0] not in names:
- utilities.append(utility)
- return utilities
-
- def getLocalUtilitiesFor(self, interface):
- """See zope.app.interfaces.services.ILocalUtilityService"""
- utilities = []
- for name in self._utilities:
- for iface, cr in self._utilities[name].getRegisteredMatching():
- if not cr or cr.active() is None:
- continue
- utility = cr.active().getComponent()
- if interface and not iface.extends(interface, 0) and \
- removeAllProxies(utility) is not interface:
- continue
- utilities.append((name, utility))
- return utilities
-
-
- def getRegisteredMatching(self, interface=None, name=None):
- """See zope.app.interfaces.services.ILocalUtilityService"""
- L = []
- for reg_name in self._utilities:
- for iface, cr in self._utilities[reg_name].getRegisteredMatching():
- if not cr:
- continue
- if interface and not iface is interface:
- continue
- if name is not None and reg_name.find(name) < 0:
- continue
- L.append((iface, reg_name, cr))
- return L
-
- def queryRegistrationsFor(self, registration, default=None):
- """zope.app.registration.interfaces.IRegistry"""
- return self.queryRegistrations(registration.name,
- registration.interface,
- default)
-
- def queryRegistrations(self, name, interface, default=None):
- """zope.app.registration.interfaces.IRegistry"""
- utilities = self._utilities.get(name)
- if utilities is None:
- return default
- stack = utilities.getRegistered(interface)
- if stack is None:
- return default
-
- return stack
-
- def createRegistrationsFor(self, registration):
- """zope.app.registration.interfaces.IRegistry"""
- return self.createRegistrations(registration.name,
- registration.interface)
-
-
- def createRegistrations(self, name, interface):
- """zope.app.interfaces.services.registration.IRegistry"""
- utilities = self._utilities.get(name)
- if utilities is None:
- utilities = ImplementorRegistry(PersistentDict())
- self._utilities[name] = utilities
-
- stack = utilities.getRegistered(interface)
- if stack is None:
- stack = RegistrationStack(self)
- utilities.register(interface, stack)
-
- return stack
-
-
-class UtilityRegistration(ComponentRegistration):
- """Utility component registration for persistent components
-
- This registration configures persistent components in packages to
- be utilities.
- """
-
- serviceType = 'Utilities'
-
- implements(IUtilityRegistration)
-
- def __init__(self, name, interface, component_path, permission=None):
- ComponentRegistration.__init__(self, component_path, permission)
- self.name = name
- self.interface = interface
-
- def usageSummary(self):
- # Override IRegistration.usageSummary()
- component = self.getComponent()
- s = self.getInterface().getName()
- if self.name:
- s += " registered as '%s'" % self.name
- s += ", implemented by %s" %component.__class__.__name__
- s += " '%s'"%self.componentPath
- return s
-
- def getInterface(self):
- # ComponentRegistration calls this when you specify a
- # permission; it needs the interface to create a security
- # proxy for the interface with the given permission.
- # XXX Smells like a dead chicken to me.
- return self.interface
-
-# XXX Pickle backward compatability
-UtilityConfiguration = UtilityRegistration
+from utility import UtilityRegistration, LocalUtilityService
=== Zope3/src/zope/app/utility/tests.py 1.5 => 1.6 ===
--- Zope3/src/zope/app/utility/tests.py:1.5 Sun Apr 11 14:16:29 2004
+++ Zope3/src/zope/app/utility/tests.py Thu Apr 15 11:29:31 2004
@@ -84,7 +84,7 @@
def setUp(self):
sm = placefulsetup.PlacefulSetup.setUp(self, site=True)
- setup.addService(sm, "Utilities",
+ setup.addService(sm, zapi.servicenames.Utilities,
utility.LocalUtilityService())
def test_queryUtility_delegates_to_global(self):
@@ -94,6 +94,12 @@
name="bob")
utility_service = getService(self.rootFolder, "Utilities")
+
+ # We changed the root (base) service. This doesn't normally
+ # occur. We have to notify the local service that the base
+ # has changes:
+ utility_service.baseChanged()
+
self.assert_(utility_service != utilityService)
self.assertEqual(utility_service.queryUtility(IFoo).foo(),
@@ -159,6 +165,12 @@
name="bob")
utilities = getService(self.rootFolder, "Utilities")
+
+ # We changed the root (base) service. This doesn't normally
+ # occur. We have to notify the local service that the base
+ # has changes:
+ utilities.baseChanged()
+
default = traverse(self.rootFolder, "++etc++site/default")
default['foo'] = Foo("local")
path = "/++etc++site/default/foo"
@@ -202,6 +214,21 @@
r = list(utilities.getRegisteredMatching())
self.assertEqual(r, [(IFoo, "bob", cr2)])
+ def test_local_overrides(self):
+ # Make sure that a local utility service can override another
+ sm1 = zapi.traverse(self.rootFolder, "++etc++site")
+ setup.addUtility(sm1, 'u1', IFoo, Foo('u1'))
+ setup.addUtility(sm1, 'u2', IFoo, Foo('u2'))
+ sm2 = self.makeSite('folder1')
+ setup.addService(sm2, zapi.servicenames.Utilities,
+ utility.LocalUtilityService())
+ setup.addUtility(sm2, 'u2', IFoo, Foo('u22'))
+
+ # Make sure we acquire:
+ self.assertEqual(zapi.getUtility(sm2, IFoo, 'u1').name, 'u1')
+
+ # Make sure we override:
+ self.assertEqual(zapi.getUtility(sm2, IFoo, 'u2').name, 'u22')
def test_suite():
return unittest.makeSuite(TestUtilityService)
More information about the Zope3-Checkins
mailing list