via the GenericBackrefExtension object.
"""
-import sqlalchemy.util as util
+import util
class SmartProperty(object):
"""Provides a property object that will communicate set/get/delete operations
create_prop method on AttributeManger, which can be overridden to provide
subclasses of SmartProperty.
"""
- def __init__(self, manager):
+ def __init__(self, manager, key, uselist):
self.manager = manager
- def attribute_registry(self):
- return self.manager
- def property(self, key, uselist):
- def set_prop(obj, value):
- self.attribute_registry().set_attribute(obj, key, value)
- def del_prop(obj):
- self.attribute_registry().delete_attribute(obj, key)
- def get_prop(obj):
- if uselist:
- return self.attribute_registry().get_list_attribute(obj, key)
- else:
- return self.attribute_registry().get_attribute(obj, key)
-
- return property(get_prop, set_prop, del_prop)
+ self.key = key
+ self.uselist = uselist
+ def __set__(self, obj, value):
+ self.manager.set_attribute(obj, self.key, value)
+ def __delete__(self, obj):
+ self.manager.delete_attribute(obj, self.key)
+ def __get__(self, obj, owner):
+ if obj is None:
+ return self
+ if self.uselist:
+ return self.manager.get_list_attribute(obj, self.key)
+ else:
+ return self.manager.get_attribute(obj, self.key)
class PropHistory(object):
"""Used by AttributeManager to track the history of a scalar attribute
upon an attribute change of value."""
pass
- def create_prop(self, key, uselist, **kwargs):
+ def create_prop(self, class_, key, uselist, **kwargs):
"""creates a scalar property object, defaulting to SmartProperty, which
will communicate change events back to this AttributeManager."""
- return SmartProperty(self).property(key, uselist)
+ return SmartProperty(self, key, uselist)
def create_list(self, obj, key, list_, **kwargs):
"""creates a history-aware list property, defaulting to a ListElement which
is a subclass of HistoryArrayList."""
return p
self.class_managed(class_)[key] = createprop
- setattr(class_, key, self.create_prop(key, uselist))
+ setattr(class_, key, self.create_prop(class_, key, uselist))
engines = property(lambda s: [t.engine for t in s.tables])
def add_property(self, key, prop):
+ """adds an additional property to this mapper. this is the same as if it were
+ specified within the 'properties' argument to the constructor. if the named
+ property already exists, this will replace it. Useful for
+ circular relationships, or overriding the parameters of auto-generated properties
+ such as backreferences."""
if sql.is_column(prop):
self.columns[key] = prop
prop = ColumnProperty(prop)
def import_instance(instance):
return get_session().import_instance(instance)
+class UOWProperty(attributes.SmartProperty):
+ def __init__(self, class_, *args, **kwargs):
+ super(UOWProperty, self).__init__(*args, **kwargs)
+ self.class_ = class_
+ property = property(lambda s:class_mapper(s.class_).props[s.key], doc="returns the MapperProperty object associated with this property")
+
class UOWListElement(attributes.ListElement):
def __init__(self, obj, key, data=None, deleteremoved=False, **kwargs):
attributes.ListElement.__init__(self, obj, key, data=data, **kwargs)
get_session(obj).register_dirty(obj)
else:
get_session(obj).register_new(obj)
+
+ def create_prop(self, class_, key, uselist, **kwargs):
+ return UOWProperty(class_, self, key, uselist)
def create_list(self, obj, key, list_, **kwargs):
return UOWListElement(obj, key, list_, **kwargs)
class DependencySorter(topological.QueueDependencySorter):
pass
-
+
def mapper(*args, **params):
return sqlalchemy.mapperlib.mapper(*args, **params)
def object_mapper(obj):
return sqlalchemy.mapperlib.object_mapper(obj)
-global_attributes = UOWAttributeManager()
+def class_mapper(class_):
+ return sqlalchemy.mapperlib.class_mapper(class_)
+global_attributes = UOWAttributeManager()
session_registry = util.ScopedRegistry(Session) # Default session registry
_sessions = weakref.WeakValueDictionary() # all referenced sessions (including user-created)
objectstore.uow().register_attribute(class_, key, uselist = self.uselist, deleteremoved = self.private, extension=self.attributeext)
def _get_direction(self):
+ """determines our 'direction', i.e. do we represent one to many, many to many, etc."""
# print self.key, repr(self.parent.table.name), repr(self.parent.primarytable.name), repr(self.foreignkey.table.name)
if self.parent.table is self.target:
if self.foreignkey.primary_key:
return (obj2, obj1)
def process_dependencies(self, task, deplist, uowcommit, delete = False):
+ """this method is called during a commit operation to synchronize data between a parent and child object.
+ it also can establish child or parent objects within the unit of work as "to be saved" or "deleted"
+ in some cases."""
#print self.mapper.table.name + " " + self.key + " " + repr(len(deplist)) + " process_dep isdelete " + repr(delete) + " direction " + repr(self.direction)
def getlist(obj, passive=True):
self.assert_result(l, User, *[{'user_id':8}])
l = m.select_by(User.c.user_name=='fred', addresses.c.email_address!='ed@bettyboop.com', user_id=9)
+
+ def testprops(self):
+ """tests the various attributes of the properties attached to classes"""
+ m = mapper(User, users, properties = {
+ 'addresses' : relation(mapper(Address, addresses))
+ })
+ self.assert_(User.addresses.property is m.props['addresses'])
def testload(self):
"""tests loading rows with a mapper and producing object instances"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
