0.4.5
=====
- orm
- - when attributes are expired on a pending instance, an
- error will not be raised when the "refresh" action
- is triggered and returns no result
-
- - fixed/covered case when using a False value as a
- polymorphic discriminator
-
- - fixed bug which was preventing synonym() attributes
- from being used with inheritance
+ - Added comparable_property(), adds query Comparator behavior
+ to regular, unmanaged Python properties
+
+ - When attributes are expired on a pending instance, an error
+ will not be raised when the "refresh" action is triggered
+ and returns no result
+
+ - Fixed/covered case when using a False value as a polymorphic
+ discriminator
+
+ - Fixed bug which was preventing synonym() attributes from
+ being used with inheritance
- Session.execute can now find binds from metadata
-
- - fixed "cascade delete" operation of dynamic relations,
- which had only been implemented for foreign-key nulling
- behavior in 0.4.2 and not actual cascading deletes
- [ticket:895]
-
+
+ - Fixed "cascade delete" operation of dynamic relations, which
+ had only been implemented for foreign-key nulling behavior
+ in 0.4.2 and not actual cascading deletes [ticket:895]
+
- extensions
- - the "synonym" function is now directly usable with
- "declarative". Pass in the decorated property using
- the "instrument" keyword argument, e.g.:
- somekey = synonym('_somekey', instrument=property(g, s))
-
+ - The "synonym" function is now directly usable with
+ "declarative". Pass in the decorated property using the
+ "instrument" keyword argument, e.g.: somekey =
+ synonym('_somekey', instrument=property(g, s))
+
0.4.4
------
- sql
"""
from sqlalchemy.orm.interfaces import PropComparator, MapperProperty
-from sqlalchemy.orm import session as sessionlib
+from sqlalchemy.orm import session as sessionlib, comparable_property
# Using the VerticalPropertyDictMixin from the base example
from dictlike import VerticalPropertyDictMixin
return '<%s %r=%r>' % (self.__class__.__name__, self.key, self.value)
-class comparable_property(MapperProperty):
- """Instruments a Python property for use in query expressions."""
-
- def __init__(self, comparator, property):
- self.property = property
- self.comparator = comparator(self)
-
- def do_init(self):
- class_ = self.parent.class_
- sessionlib.register_attribute(class_, self.key, uselist=False,
- proxy_property=self.property,
- useobject=False,
- comparator=self.comparator)
-
- def setup(self, querycontext, **kwargs):
- pass
-
- def create_row_processor(self, selectcontext, mapper, row):
- return (None, None, None)
-
-
if __name__ == '__main__':
from sqlalchemy import *
from sqlalchemy.orm import mapper, relation, create_session
from sqlalchemy import util as sautil
from sqlalchemy.orm.mapper import Mapper, object_mapper, class_mapper, _mapper_registry
from sqlalchemy.orm.interfaces import MapperExtension, EXT_CONTINUE, EXT_STOP, EXT_PASS, ExtensionOption, PropComparator
-from sqlalchemy.orm.properties import SynonymProperty, PropertyLoader, ColumnProperty, CompositeProperty, BackRef
+from sqlalchemy.orm.properties import SynonymProperty, ComparableProperty, PropertyLoader, ColumnProperty, CompositeProperty, BackRef
from sqlalchemy.orm import mapper as mapperlib
from sqlalchemy.orm import strategies
from sqlalchemy.orm.query import Query
'undefer', 'undefer_group', 'extension', 'mapper', 'clear_mappers',
'compile_mappers', 'class_mapper', 'object_mapper', 'sessionmaker',
'scoped_session', 'dynamic_loader', 'MapperExtension',
- 'polymorphic_union',
+ 'polymorphic_union', 'comparable_property',
'create_session', 'synonym', 'contains_alias', 'Query',
'contains_eager', 'EXT_CONTINUE', 'EXT_STOP', 'EXT_PASS',
'object_session', 'PropComparator' ]
return SynonymProperty(name, map_column=map_column, instrument=instrument)
+def comparable_property(comparator_factory, descriptor=None):
+ """Provide query semantics for an unmanaged attribute.
+
+ Allows a regular Python @property (descriptor) to be used in Queries and
+ SQL constructs like a managed attribute. comparable_property wraps a
+ descriptor with a proxy that directs operator overrides such as ==
+ (__eq__) to the supplied comparator but proxies everything else through
+ to the original descriptor.
+
+ class MyClass(object):
+ @property
+ def myprop(self):
+ return 'foo'
+
+ class MyComparator(sqlalchemy.orm.interfaces.PropComparator):
+ def __eq__(self, other):
+ ....
+
+ mapper(MyClass, mytable, properties=dict(
+ 'myprop': comparable_property(MyComparator)))
+
+ Used with the ``properties`` dictionary sent to [sqlalchemy.orm#mapper()].
+
+ comparator_factory
+ A PropComparator subclass or factory that defines operator behavior
+ for this property.
+
+ descriptor
+
+ Optional when used in a ``properties={}`` declaration. The Python
+ descriptor or property to layer comparison behavior on top of.
+
+ The like-named descriptor will be automatically retreived from the
+ mapped class if left blank in a ``properties`` declaration.
+ """
+ return ComparableProperty(comparator_factory, descriptor)
+
def compile_mappers():
"""Compile all mappers that have been defined.
# lock used to synchronize the "mapper compile" step
_COMPILE_MUTEX = util.threading.Lock()
-# initialize these two lazily
+# initialize these lazily
ColumnProperty = None
SynonymProperty = None
+ComparableProperty = None
+
class Mapper(object):
"""Define the correlation of class attributes to database table
_equivalent_columns = property(_equivalent_columns)
class _CompileOnAttr(PropComparator):
- """placeholder class attribute which fires mapper compilation on access"""
+ """A placeholder descriptor which triggers compilation on access."""
def __init__(self, class_, key):
self.class_ = class_
if not key in self.mapped_table.c:
raise exceptions.ArgumentError("Can't compile synonym '%s': no column on table '%s' named '%s'" % (prop.name, self.mapped_table.description, key))
self._compile_property(prop.name, ColumnProperty(self.mapped_table.c[key]), init=init, setparent=setparent)
+ elif isinstance(prop, ComparableProperty) and setparent:
+ # refactor me
+ if prop.descriptor is None:
+ prop.descriptor = getattr(self.class_, key, None)
+ if isinstance(prop.descriptor, Mapper._CompileOnAttr):
+ prop.descriptor = object.__getattribute__(prop.descriptor,
+ 'existing_prop')
self.__props[key] = prop
if setparent:
import weakref
__all__ = ('ColumnProperty', 'CompositeProperty', 'SynonymProperty',
- 'PropertyLoader', 'BackRef')
+ 'ComparableProperty', 'PropertyLoader', 'BackRef')
class ColumnProperty(StrategizedProperty):
pass
SynonymProperty.logger = logging.class_logger(SynonymProperty)
+
+class ComparableProperty(MapperProperty):
+ """Instruments a Python property for use in query expressions."""
+
+ def __init__(self, comparator_factory, descriptor=None):
+ self.descriptor = descriptor
+ self.comparator = comparator_factory(self)
+
+ def do_init(self):
+ """Set up a proxy to the unmanaged descriptor."""
+
+ class_ = self.parent.class_
+ # refactor me
+ sessionlib.register_attribute(class_, self.key, uselist=False,
+ proxy_property=self.descriptor,
+ useobject=False,
+ comparator=self.comparator)
+
+ def setup(self, querycontext, **kwargs):
+ pass
+
+ def create_row_processor(self, selectcontext, mapper, row):
+ return (None, None, None)
+
+
class PropertyLoader(StrategizedProperty):
"""Describes an object property that holds a single item or list
of items that correspond to a related database table.
mapper.ColumnProperty = ColumnProperty
mapper.SynonymProperty = SynonymProperty
+mapper.ComparableProperty = ComparableProperty
def test_add_property(self):
assert_col = []
+
class User(object):
def _get_user_name(self):
assert_col.append(('get', self._user_name))
self._user_name = name
user_name = property(_get_user_name, _set_user_name)
+ def _uc_user_name(self):
+ if self._user_name is None:
+ return None
+ return self._user_name.upper()
+ uc_user_name = property(_uc_user_name)
+ uc_user_name2 = property(_uc_user_name)
+
m = mapper(User, users)
mapper(Address, addresses)
+
+ class UCComparator(PropComparator):
+ def __eq__(self, other):
+ cls = self.prop.parent.class_
+ col = getattr(cls, 'user_name')
+ if other is None:
+ return col == None
+ else:
+ return func.upper(col) == func.upper(other)
+
m.add_property('_user_name', deferred(users.c.user_name))
m.add_property('user_name', synonym('_user_name'))
m.add_property('addresses', relation(Address))
+ m.add_property('uc_user_name', comparable_property(UCComparator))
+ m.add_property('uc_user_name2', comparable_property(
+ UCComparator, User.uc_user_name2))
sess = create_session(transactional=True)
assert sess.query(User).get(7)
def go():
self.assert_result([u], User, user_address_result[0])
assert u.user_name == 'jack'
+ assert u.uc_user_name == 'JACK'
+ assert u.uc_user_name2 == 'JACK'
assert assert_col == [('get', 'jack')], str(assert_col)
self.assert_sql_count(testing.db, go, 2)
assert u.user_name == 'foo'
assert assert_col == [('get', 'jack'), ('set', 'foo'), ('get', 'foo')]
+ def test_comparable(self):
+ class extendedproperty(property):
+ attribute = 123
+ def __getitem__(self, key):
+ return 'value'
+
+ class UCComparator(PropComparator):
+ def __eq__(self, other):
+ cls = self.prop.parent.class_
+ col = getattr(cls, 'user_name')
+ if other is None:
+ return col == None
+ else:
+ return func.upper(col) == func.upper(other)
+
+ def map_(with_explicit_property):
+ class User(object):
+ @extendedproperty
+ def uc_user_name(self):
+ if self.user_name is None:
+ return None
+ return self.user_name.upper()
+ if with_explicit_property:
+ args = (UCComparator, User.uc_user_name)
+ else:
+ args = (UCComparator,)
+
+ mapper(User, users, properties=dict(
+ uc_user_name = comparable_property(*args)))
+ return User
+
+ for User in (map_(True), map_(False)):
+ sess = create_session()
+ sess.begin()
+ q = sess.query(User)
+
+ assert hasattr(User, 'user_name')
+ assert hasattr(User, 'uc_user_name')
+
+ # test compile
+ assert not isinstance(User.uc_user_name == 'jack', bool)
+ u = q.filter(User.uc_user_name=='JACK').one()
+
+ assert u.uc_user_name == "JACK"
+ assert u not in sess.dirty
+
+ u.user_name = "some user name"
+ assert u.user_name == "some user name"
+ assert u in sess.dirty
+ assert u.uc_user_name == "SOME USER NAME"
+
+ sess.flush()
+ sess.clear()
+
+ q = sess.query(User)
+ u2 = q.filter(User.user_name=='some user name').one()
+ u3 = q.filter(User.uc_user_name=='SOME USER NAME').one()
+
+ assert u2 is u3
+
+ assert User.uc_user_name.attribute == 123
+ assert User.uc_user_name['key'] == 'value'
+ sess.rollback()
+
class OptionsTest(MapperSuperTest):
@testing.fails_on('maxdb')
def test_synonymoptions(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
