[z3-checkins] r36300 - in z3/sqlos/trunk/src/sqlos: . container
kobold at codespeak.net
kobold at codespeak.net
Mon Jan 8 18:08:01 CET 2007
Author: kobold
Date: Mon Jan 8 18:07:59 2007
New Revision: 36300
Added:
z3/sqlos/trunk/src/sqlos/container/
z3/sqlos/trunk/src/sqlos/container/__init__.py
- copied unchanged from r36286, z3/sqlos/trunk/src/sqlos/container.py
Removed:
z3/sqlos/trunk/src/sqlos/container.py
Log:
Transformed container.py module into a package.
Deleted: /z3/sqlos/trunk/src/sqlos/container.py
==============================================================================
--- /z3/sqlos/trunk/src/sqlos/container.py Mon Jan 8 18:07:59 2007
+++ (empty file)
@@ -1,258 +0,0 @@
-##############################################################################
-#
-# Copyright (c) 2004 Enfold Systems LLC. All rights reserved.
-# Copyright (c) 2005-2006 Brian Sutherland. All rights reserved.
-#
-# This software is distributed under the terms of the Zope Public
-# License (ZPL) v2.1. See COPYING.txt for more information.
-#
-##############################################################################
-"""
-$Id$
-"""
-
-import random
-
-from sqlobject import *
-from persistent import Persistent
-from zope.interface import implements
-from zope.component import IFactory
-import zope.component
-from zope.app.container.interfaces import IContained
-from zope.app.container.contained import ContainedProxy
-from zope.app.container.contained import Contained
-from zope.app.container.contained import NameChooser
-from zope.app.container.constraints import checkFactory
-from zope.location.interfaces import ILocation
-from zope.proxy import sameProxiedObjects
-from zope.exceptions.interfaces import UserError
-
-from sqlos.interfaces import ISQLObject, ISQLObjectIsolated, IISQLObject
-from sqlos.interfaces.container import ISQLObjectContainer
-from sqlos.interfaces.container import IIsolatedSQLContainer
-
-def contained(obj, parent=None, name=None):
- """An implementation of zope.app.container.contained.contained
- that doesn't generate events, for internal use.
- """
- if (parent is None):
- raise TypeError, 'Must provide a parent'
-
- if not IContained.providedBy(obj):
- if ILocation.providedBy(obj):
- directlyProvides(obj, IContained, directlyProvidedBy(obj))
- else:
- obj = ContainedProxy(obj)
-
- oldparent = obj.__parent__
- oldname = obj.__name__
-
- if (oldparent is None) or not (oldparent is parent
- or sameProxiedObjects(oldparent, parent)):
- obj.__parent__ = parent
-
- if oldname != name and name is not None:
- obj.__name__ = unicode(name)
-
- return obj
-
-
-class SQLObjectNameChooser(NameChooser):
- # XXX: This needs unit tests...
-
- def chooseName(self, name, obj):
- if ISQLObject.providedBy(obj):
- # Look for the SQLObject class our object is from, so get all
- # allowed factories
- for name, factory in zope.component.getFactoriesFor(ISQLObject):
- if checkFactory(self.context, None, factory):
- # get the sqlobject class
- utility = zope.component.queryUtility(IISQLObject, name)
- if utility is None:
- continue
- if obj.sqlmeta.table == utility.sqlmeta.table:
- # if the tables names are the same, we assume that the
- # sqlobject is an instance of that class, perhaps that
- # is wrong
- return "%s.%s" % (name, obj.id)
- raise UserError("Cannot find a name") # XXX better message, i18n?
-
-
-class SQLObjectContainer(Persistent, Contained):
-
- implements(ISQLObjectContainer)
-
- def __init__(self):
- pass
-
- def _getAllowedIISQLObjectUtilities(self):
- for name, factory in zope.component.getFactoriesFor(ISQLObject):
- if checkFactory(self, None, factory):
- utility = zope.component.queryUtility(IISQLObject, name)
- # Someone might have registered a factory implementing
- # IISQLObject using <zope:factory> for whatever reason.
- # in this case queryUtility returns None and we can just
- # ignore it
- if utility is not None:
- yield name, utility
-
- def keys(self):
- """ Return a sequence-like object containing the names
- associated with the objects that appear in the folder
- """
- for name, obj in self.items():
- yield name
-
- def __iter__(self):
- return iter(self.keys())
-
- def values(self):
- """ Return a sequence-like object containing the objects that
- appear in the folder.
- """
- for name, obj in self.items():
- yield contained(obj, parent=self, name=name)
-
- def items(self):
- """Return a sequence-like object containing tuples of the form
- (name, object) for the objects that appear in the folder.
- """
- for utility_name, utility in self._getAllowedIISQLObjectUtilities():
- for obj in utility.select():
- name = '%s.%s' % (utility_name, obj.id)
- yield (name, contained(obj, parent=self, name=name))
-
- def __getitem__(self, name):
- """Return the named object.
-
- If the object is not found a KeyError is raised.
-
- lets get a container:
-
- >>> c = SQLObjectContainer()
-
- And make sure it doesn't bork on non-string values:
-
- >>> c[None]
- Traceback (most recent call last):
- ...
- KeyError: ...
- >>> c[object()]
- Traceback (most recent call last):
- ...
- KeyError: ...
- """
- if not isinstance(name, basestring):
- raise KeyError, "%s is not a string" % name
- try:
- parts = name.split('.')
- id = parts[-1]
- factoryName = '.'.join(parts[:-1])
- except ValueError:
- raise KeyError, name
-
- for utility_name, utility in self._getAllowedIISQLObjectUtilities():
- if factoryName != utility_name:
- continue
- try:
- obj = utility.get(utility.sqlmeta.idType(id))
- return contained(obj, parent=self, name=name)
- except (SQLObjectNotFound, ValueError):
- # SQlObject raises ValueError if the key is not correct
- raise KeyError, name
- raise KeyError, name
-
- def get(self, name, default=None):
- """Return the named object, or the value of the default
- argument if given and the named object is not found.
- If no default is given and the object is not found a
- KeyError is raised.
- """
- try:
- return contained(self[name], parent=self, name=name)
- except KeyError:
- return default
-
- def __contains__(self, name):
- """Return true if the named object appears in the folder."""
- return self.get(name, None) is not None
-
- def __len__(self):
- """Return the number of objects in the folder."""
- i = 0
- for utility_name, utility in self._getAllowedIISQLObjectUtilities():
- i += utility.select().count() # optimal, does not get all objects
- return i
-
- def __delitem__(self, name):
- """Delete the named object from the container.
-
- Raises a KeyError if the object is not found.
- """
- obj = self[name]
- obj.destroySelf()
-
- def __setitem__(self, name, content):
- return name
-
-
-class SQLIsolatedContainer(SQLObjectContainer):
-
- implements(IIsolatedSQLContainer)
-
- _container_id = None
-
- def _getAllowedIISQLObjectUtilities(self):
- # Ignore all utilities not implementing ISQLObjectIsolated
- for name, utility in SQLObjectContainer._getAllowedIISQLObjectUtilities(self):
- if ISQLObjectIsolated.implementedBy(utility):
- yield name, utility
-
- def _getContainerId(self):
- if self._container_id is None:
- self._container_id = str(random.random()) # TODO better generation?
- return self._container_id
- def _setContainerId(self, value):
- self._container_id = value
- container_id = property(_getContainerId, _setContainerId)
-
- def __len__(self):
- """Return the number of objects in the folder."""
- i = 0
- for name, utility in self._getAllowedIISQLObjectUtilities():
- i += utility.countByDomain(self.container_id)
- return i
-
- def __delitem__(self, name):
- """Delete the named object from the container.
-
- Raises a KeyError if the object is not found.
- """
- obj = self[name]
- domains = [d for d in obj.domains if d != self.container_id]
- if not domains:
- obj.destroySelf()
- return
- obj.domains = tuple(domains)
-
- def __setitem__(self, name, content):
- domains = content.domains
- if self.container_id not in domains:
- content.domains = domains + (self.container_id, )
- return name
-
- def items(self):
- """Return a sequence-like object containing tuples of the form
- (name, object) for the objects that appear in the folder.
- """
- for utility_name, utility in self._getAllowedIISQLObjectUtilities():
- for obj in utility.selectByDomain(self.container_id):
- name = '%s.%s' % (utility_name, obj.id)
- yield (name, contained(obj, parent=self, name=name))
-
- def __getitem__(self, name):
- obj = super(SQLIsolatedContainer, self).__getitem__(name)
- if hasattr(obj, 'domains'):
- if self.container_id in obj.domains:
- return contained(obj, parent=self, name=name)
- raise KeyError, name
More information about the z3-checkins
mailing list