/usr/share/pyshared/axiom/dependency.py is in python-axiom 0.7.1-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 | # Copright 2008 Divmod, Inc. See LICENSE file for details.
# -*- test-case-name: axiom.test.test_dependency -*-
"""
A dependency management system for items.
"""
import sys, itertools
from zope.interface.advice import addClassAdvisor
from epsilon.structlike import record
from axiom.item import Item
from axiom.attributes import reference, boolean, AND
from axiom.errors import ItemNotFound, DependencyError, UnsatisfiedRequirement
#There is probably a cleaner way to do this.
_globalDependencyMap = {}
def dependentsOf(cls):
deps = _globalDependencyMap.get(cls, None)
if deps is None:
return []
else:
return [d[0] for d in deps]
##Totally ripping off z.i
def dependsOn(itemType, itemCustomizer=None, doc='',
indexed=True, whenDeleted=reference.NULLIFY):
"""
This function behaves like L{axiom.attributes.reference} but with
an extra behaviour: when this item is installed (via
L{axiom.dependency.installOn} on a target item, the
type named here will be instantiated and installed on the target
as well.
For example::
class Foo(Item):
counter = integer()
thingIDependOn = dependsOn(Baz, lambda baz: baz.setup())
@param itemType: The Item class to instantiate and install.
@param itemCustomizer: A callable that accepts the item installed
as a dependency as its first argument. It will be called only if
an item is created to satisfy this dependency.
@return: An L{axiom.attributes.reference} instance.
"""
frame = sys._getframe(1)
locals = frame.f_locals
# Try to make sure we were called from a class def.
if (locals is frame.f_globals) or ('__module__' not in locals):
raise TypeError("dependsOn can be used only from a class definition.")
ref = reference(reftype=itemType, doc=doc, indexed=indexed, allowNone=True,
whenDeleted=whenDeleted)
if "__dependsOn_advice_data__" not in locals:
addClassAdvisor(_dependsOn_advice)
locals.setdefault('__dependsOn_advice_data__', []).append(
(itemType, itemCustomizer, ref))
return ref
def _dependsOn_advice(cls):
if cls in _globalDependencyMap:
print "Double advising of %s. dependency map from first time: %s" % (
cls, _globalDependencyMap[cls])
#bail if we end up here twice, somehow
return cls
for itemType, itemCustomizer, ref in cls.__dict__[
'__dependsOn_advice_data__']:
classDependsOn(cls, itemType, itemCustomizer, ref)
del cls.__dependsOn_advice_data__
return cls
def classDependsOn(cls, itemType, itemCustomizer, ref):
_globalDependencyMap.setdefault(cls, []).append(
(itemType, itemCustomizer, ref))
class _DependencyConnector(Item):
"""
I am a connector between installed items and their targets.
"""
installee = reference(doc="The item installed.")
target = reference(doc="The item installed upon.")
explicitlyInstalled = boolean(doc="Whether this item was installed"
"explicitly (and thus whether or not it"
"should be automatically uninstalled when"
"nothing depends on it)")
def installOn(self, target):
"""
Install this object on the target along with any powerup
interfaces it declares. Also track that the object now depends on
the target, and the object was explicitly installed (and therefore
should not be uninstalled by subsequent uninstallation operations
unless it is explicitly removed).
"""
_installOn(self, target, True)
def _installOn(self, target, __explicitlyInstalled=False):
depBlob = _globalDependencyMap.get(self.__class__, [])
dependencies, itemCustomizers, refs = (map(list, zip(*depBlob))
or ([], [], []))
#See if any of our dependencies have been installed already
for dc in self.store.query(_DependencyConnector,
_DependencyConnector.target == target):
if dc.installee.__class__ in dependencies:
i = dependencies.index(dc.installee.__class__)
refs[i].__set__(self, dc.installee)
del dependencies[i], itemCustomizers[i], refs[i]
if (dc.installee.__class__ == self.__class__
and self.__class__ in set(
itertools.chain([blob[0][0] for blob in
_globalDependencyMap.values()]))):
#Somebody got here before we did... let's punt
raise DependencyError("An instance of %r is already "
"installed on %r." % (self.__class__,
target))
#The rest we'll install
for i, cls in enumerate(dependencies):
it = cls(store=self.store)
if itemCustomizers[i] is not None:
itemCustomizers[i](it)
_installOn(it, target, False)
refs[i].__set__(self, it)
#And now the connector for our own dependency.
dc = self.store.findUnique(
_DependencyConnector,
AND(_DependencyConnector.target==target,
_DependencyConnector.installee==self,
_DependencyConnector.explicitlyInstalled==__explicitlyInstalled),
None)
assert dc is None, "Dependency connector already exists, wtf are you doing?"
_DependencyConnector(store=self.store, target=target,
installee=self,
explicitlyInstalled=__explicitlyInstalled)
target.powerUp(self)
callback = getattr(self, "installed", None)
if callback is not None:
callback()
def uninstallFrom(self, target):
"""
Remove this object from the target, as well as any dependencies
that it automatically installed which were not explicitly
"pinned" by calling "install", and raising an exception if
anything still depends on this.
"""
#did this class powerup on any interfaces? powerdown if so.
target.powerDown(self)
for dc in self.store.query(_DependencyConnector,
_DependencyConnector.target==target):
if dc.installee is self:
dc.deleteFromStore()
for item in installedUniqueRequirements(self, target):
uninstallFrom(item, target)
callback = getattr(self, "uninstalled", None)
if callback is not None:
callback()
def installedOn(self):
"""
If this item is installed on another item, return the install
target. Otherwise return None.
"""
try:
return self.store.findUnique(_DependencyConnector,
_DependencyConnector.installee == self
).target
except ItemNotFound:
return None
def installedDependents(self, target):
"""
Return an iterable of things installed on the target that
require this item.
"""
for dc in self.store.query(_DependencyConnector,
_DependencyConnector.target == target):
depends = dependentsOf(dc.installee.__class__)
if self.__class__ in depends:
yield dc.installee
def installedUniqueRequirements(self, target):
"""
Return an iterable of things installed on the target that this item
requires and are not required by anything else.
"""
myDepends = dependentsOf(self.__class__)
#XXX optimize?
for dc in self.store.query(_DependencyConnector,
_DependencyConnector.target==target):
if dc.installee is self:
#we're checking all the others not ourself
continue
depends = dependentsOf(dc.installee.__class__)
if self.__class__ in depends:
raise DependencyError(
"%r cannot be uninstalled from %r, "
"%r still depends on it" % (self, target, dc.installee))
for cls in myDepends[:]:
#If one of my dependencies is required by somebody
#else, leave it alone
if cls in depends:
myDepends.remove(cls)
for dc in self.store.query(_DependencyConnector,
_DependencyConnector.target==target):
if (dc.installee.__class__ in myDepends
and not dc.explicitlyInstalled):
yield dc.installee
def installedRequirements(self, target):
"""
Return an iterable of things installed on the target that this
item requires.
"""
myDepends = dependentsOf(self.__class__)
for dc in self.store.query(_DependencyConnector,
_DependencyConnector.target == target):
if dc.installee.__class__ in myDepends:
yield dc.installee
def onlyInstallPowerups(self, target):
"""
Deprecated - L{Item.powerUp} now has this functionality.
"""
target.powerUp(self)
class requiresFromSite(
record('powerupInterface defaultFactory siteDefaultFactory',
defaultFactory=None,
siteDefaultFactory=None)):
"""
A read-only descriptor that will return the site store's powerup for a
given item.
@ivar powerupInterface: an L{Interface} describing the powerup that the
site store should be adapted to.
@ivar defaultFactory: a 1-argument callable that takes the site store and
returns a value for this descriptor. This is invoked in cases where the
site store does not provide a default factory of its own, and this
descriptor is retrieved from an item in a store with a parent.
@ivar siteDefaultFactory: a 1-argument callable that takes the site store
and returns a value for this descriptor. This is invoked in cases where
this descriptor is retrieved from an item in a store without a parent.
"""
def _invokeFactory(self, defaultFactory, siteStore):
if defaultFactory is None:
raise UnsatisfiedRequirement()
return defaultFactory(siteStore)
def __get__(self, oself, type=None):
"""
Retrieve the value of this dependency from the site store.
"""
siteStore = oself.store.parent
if siteStore is not None:
pi = self.powerupInterface(siteStore, None)
if pi is None:
pi = self._invokeFactory(self.defaultFactory, siteStore)
else:
pi = self._invokeFactory(self.siteDefaultFactory, oself.store)
return pi
|