- orm
+ - new synonym() behavior: an attribute will be placed on the mapped
+ class, if one does not exist already, in all cases. if a property
+ already exists on the class, the synonym will decorate the property
+ with the appropriate comparison operators so that it can be used in in
+ column expressions just like any other mapped attribute (i.e. usable in
+ filter(), etc.) the "proxy=True" flag is deprecated and no longer means
+ anything. Additionally, the flag "map_column=True" will automatically
+ generate a ColumnProperty corresponding to the name of the synonym,
+ i.e.: 'somename':synonym('_somename', map_column=True) will map the
+ column named 'somename' to the attribute '_somename'. See the example
+ in the mapper docs. [ticket:801]
+
- fixed endless loop issue when using lazy="dynamic" on both
sides of a bi-directional relationship [ticket:872]
)
})
-#### Overriding Attribute Behavior {@name=overriding}
+#### Overriding Attribute Behavior with Synonyms {@name=overriding}
-A common request is the ability to create custom class properties that override the behavior of setting/getting an attribute. You accomplish this using normal Python `property` constructs:
+A common request is the ability to create custom class properties that override the behavior of setting/getting an attribute. As of 0.4.2, the `synonym()` construct provides an easy way to do this in conjunction with a normal Python `property` constructs. Below, we re-map the `email` column of our mapped table to a custom attribute setter/getter, mapping the actual column to the property named `_email`:
{python}
class MyAddress(object):
email = property(_get_email, _set_email)
mapper(MyAddress, addresses_table, properties = {
- # map the '_email' attribute to the "email" column
- # on the table
- '_email': addresses_table.c.email
+ 'email':synonym('_email', map_column=True)
})
-To have your custom `email` property be recognized by keyword-based `Query` functions such as `filter_by()`, place a `synonym` on your mapper:
+The `email` attribute is now usable in the same way as any other mapped attribute, including filter expressions, get/set operations, etc.:
{python}
- mapper(MyAddress, addresses_table, properties = {
- '_email': addresses_table.c.email
-
- 'email':synonym('_email')
- })
-
- # use the synonym in a query
- result = session.query(MyAddress).filter_by(email='john@smith.com')
-
-Synonym strategies such as the above can be easily automated, such as this example which specifies all columns and synonyms explicitly:
+ address = sess.query(MyAddress).filter(MyAddress.email == 'some address').one()
- {python}
- mapper(MyAddress, addresses_table, properties = dict(
- [('_'+col.key, col) for col in addresses_table.c] +
- [(col.key, synonym('_'+col.key)) for col in addresses_table.c]
- ))
-
-The `column_prefix` option can also help with the above scenario by setting up the columns automatically with a prefix:
+ address.email = 'some other address'
+ sess.flush()
+
+ q = sess.query(MyAddress).filter_by(email='some other address')
- {python}
- mapper(MyAddress, addresses_table, column_prefix='_', properties = dict(
- [(col.key, synonym('_'+col.key)) for col in addresses_table.c]
- ))
+If the mapped class does not provide a property, the `synonym()` construct will create a default getter/setter object automatically.
#### Composite Column Types {@name=composite}
from sqlalchemy import util as sautil
from sqlalchemy.orm.mapper import Mapper, object_mapper, class_mapper, mapper_registry
-from sqlalchemy.orm.interfaces import SynonymProperty, MapperExtension, EXT_CONTINUE, EXT_STOP, EXT_PASS, ExtensionOption, PropComparator
-from sqlalchemy.orm.properties import PropertyLoader, ColumnProperty, CompositeProperty, BackRef
+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 import mapper as mapperlib
from sqlalchemy.orm import strategies
from sqlalchemy.orm.query import Query
return Mapper(class_, local_table, *args, **params)
-def synonym(name, proxy=False):
- """Set up `name` as a synonym to another ``MapperProperty``.
+def synonym(name, map_column=False, proxy=False):
+ """Set up `name` as a synonym to another mapped property.
- Used with the `properties` dictionary sent to ``mapper()``.
+ Used with the ``properties`` dictionary sent to [sqlalchemy.orm#mapper()].
+
+ Any existing attributes on the class which map the key name sent
+ to the ``properties`` dictionary will be used by the synonym to
+ provide instance-attribute behavior (that is, any Python property object,
+ provided by the ``property`` builtin or providing a ``__get__()``,
+ ``__set__()`` and ``__del__()`` method). If no name exists for the key,
+ the ``synonym()`` creates a default getter/setter object automatically
+ and applies it to the class.
+
+ `name` refers to the name of the existing mapped property, which
+ can be any other ``MapperProperty`` including column-based
+ properties and relations.
+
+ if `map_column` is ``True``, an additional ``ColumnProperty``
+ is created on the mapper automatically, using the synonym's
+ name as the keyname of the property, and the keyname of this ``synonym()``
+ as the name of the column to map. For example, if a table has a column
+ named ``status``::
+
+ class MyClass(object):
+ def _get_status(self):
+ return self._status
+ def _set_status(self, value):
+ self._status = value
+ status = property(_get_status, _set_status)
+
+ mapper(MyClass, sometable, properties={
+ "status":synonym("_status", map_column=True)
+ })
+
+ The column named ``status`` will be mapped to the attribute named ``_status``,
+ and the ``status`` attribute on ``MyClass`` will be used to proxy access to the
+ column-based attribute.
+
+ The `proxy` keyword argument is deprecated and currently does nothing; synonyms
+ now always establish an attribute getter/setter funciton if one is not already available.
"""
- return SynonymProperty(name, proxy=proxy)
+ return SynonymProperty(name, map_column=map_column)
def compile_mappers():
"""Compile all mappers that have been defined.
return class_mapper(self.impl.class_).get_property(self.impl.key)
property = property(_property, doc="the MapperProperty object associated with this attribute")
+class ProxiedAttribute(InstrumentedAttribute):
+ class ProxyImpl(object):
+ def __init__(self, key):
+ self.key = key
+ def commit_to_state(self, state, value=NO_VALUE):
+ pass
+
+ def __init__(self, key, user_prop, comparator=None):
+ self.user_prop = user_prop
+ self.comparator = comparator
+ self.key = key
+ self.impl = ProxiedAttribute.ProxyImpl(key)
+ def __get__(self, obj, owner):
+ if obj is None:
+ self.user_prop.__get__(obj, owner)
+ return self
+ return self.user_prop.__get__(obj, owner)
+ def __set__(self, obj, value):
+ return self.user_prop.__set__(obj, value)
+ def __delete__(self, obj):
+ return self.user_prop.__delete__(obj)
+
+
+
class AttributeImpl(object):
"""internal implementation for instrumented attributes."""
if '_sa_attrs' in class_.__dict__:
delattr(class_, '_sa_attrs')
-def register_attribute(class_, key, uselist, useobject, callable_=None, **kwargs):
+def register_attribute(class_, key, uselist, useobject, callable_=None, proxy_property=None, **kwargs):
if not '_sa_attrs' in class_.__dict__:
class_._sa_attrs = []
# TODO: possibly have InstrumentedAttribute check "entity_name" when searching for impl.
# raise an error if two attrs attached simultaneously otherwise
return
-
- inst = InstrumentedAttribute(_create_prop(class_, key, uselist, callable_, useobject=useobject,
+
+ if proxy_property:
+ inst = ProxiedAttribute(key, proxy_property, comparator=comparator)
+ else:
+ inst = InstrumentedAttribute(_create_prop(class_, key, uselist, callable_, useobject=useobject,
typecallable=typecallable, **kwargs), comparator=comparator)
setattr(class_, key, inst)
__all__ = ['EXT_CONTINUE', 'EXT_STOP', 'EXT_PASS', 'MapperExtension',
'MapperProperty', 'PropComparator', 'StrategizedProperty',
'build_path', 'MapperOption',
- 'ExtensionOption', 'SynonymProperty', 'PropertyOption',
+ 'ExtensionOption', 'PropertyOption',
'AttributeExtension', 'StrategizedOption', 'LoaderStrategy' ]
EXT_CONTINUE = EXT_PASS = object()
query._extension = query._extension.copy()
query._extension.insert(self.ext)
-class SynonymProperty(MapperProperty):
- def __init__(self, name, proxy=False):
- self.name = name
- self.proxy = proxy
-
- def setup(self, querycontext, **kwargs):
- pass
-
- def create_row_processor(self, selectcontext, mapper, row):
- return (None, None, None)
-
- def do_init(self):
- if not self.proxy:
- return
- class SynonymProp(object):
- def __set__(s, obj, value):
- setattr(obj, self.name, value)
- def __delete__(s, obj):
- delattr(obj, self.name)
- def __get__(s, obj, owner):
- if obj is None:
- return s
- return getattr(obj, self.name)
- setattr(self.parent.class_, self.key, SynonymProp())
-
- def merge(self, session, source, dest, _recursive):
- pass
class PropertyOption(MapperOption):
"""A MapperOption that is applied to a property off the mapper or
from sqlalchemy.orm import util as mapperutil
from sqlalchemy.orm.util import ExtensionCarrier, create_row_adapter
from sqlalchemy.orm import sync, attributes
-from sqlalchemy.orm.interfaces import MapperProperty, EXT_CONTINUE, SynonymProperty, PropComparator
+from sqlalchemy.orm.interfaces import MapperProperty, EXT_CONTINUE, PropComparator
deferred_load = None
__all__ = ['Mapper', 'class_mapper', 'object_mapper', 'mapper_registry']
# initialize these two lazily
ColumnProperty = None
+SynonymProperty = None
class Mapper(object):
"""Define the correlation of class attributes to database table
def __init__(self, class_, key):
self.class_ = class_
self.key = key
+
def __getattribute__(self, key):
cls = object.__getattribute__(self, 'class_')
clskey = object.__getattribute__(self, 'key')
# table columns mapped to lists of MapperProperty objects
# using a list allows a single column to be defined as
# populating multiple object attributes
- self._columntoproperty = {} #mapperutil.TranslatingDict(self.mapped_table)
+ self._columntoproperty = {}
# load custom properties
if self._init_properties is not None:
for col in prop.columns:
for col in col.proxy_set:
self._columntoproperty[col] = prop
-
+ elif isinstance(prop, SynonymProperty):
+ prop.instrument = getattr(self.class_, key, None)
+ if prop.map_column:
+ if not key in self.select_table.c:
+ raise exceptions.ArgumentError("Can't compile synonym '%s': no column on table '%s' named '%s'" % (prop.name, self.select_table.description, key))
+ self._compile_property(prop.name, ColumnProperty(self.select_table.c[key]), init=init, setparent=setparent)
self.__props[key] = prop
if setparent:
prop.set_parent(self)
- # TODO: centralize _CompileOnAttr logic, move into MapperProperty classes
- if (not isinstance(prop, SynonymProperty) or prop.proxy) and not self.non_primary and not hasattr(self.class_, key):
+ if not self.non_primary:
setattr(self.class_, key, Mapper._CompileOnAttr(self.class_, key))
if init:
from sqlalchemy.orm import session as sessionlib
from sqlalchemy.orm import util as mapperutil
import operator
-from sqlalchemy.orm.interfaces import StrategizedProperty, PropComparator
+from sqlalchemy.orm.interfaces import StrategizedProperty, PropComparator, MapperProperty
from sqlalchemy.exceptions import ArgumentError
-__all__ = ['ColumnProperty', 'CompositeProperty', 'PropertyLoader', 'BackRef']
+__all__ = ['ColumnProperty', 'CompositeProperty', 'SynonymProperty', 'PropertyLoader', 'BackRef']
class ColumnProperty(StrategizedProperty):
"""Describes an object attribute that corresponds to a table column."""
zip(self.prop.columns,
other.__composite_values__())])
+class SynonymProperty(MapperProperty):
+ def __init__(self, name, map_column=None):
+ self.name = name
+ self.map_column=map_column
+ self.instrument = None
+
+ def setup(self, querycontext, **kwargs):
+ pass
+
+ def create_row_processor(self, selectcontext, mapper, row):
+ return (None, None, None)
+
+ def do_init(self):
+ class_ = self.parent.class_
+ aliased_property = self.parent.get_property(self.key, resolve_synonyms=True)
+ self.logger.info("register managed attribute %s on class %s" % (self.key, class_.__name__))
+ if self.instrument is None:
+ class SynonymProp(object):
+ def __set__(s, obj, value):
+ setattr(obj, self.name, value)
+ def __delete__(s, obj):
+ delattr(obj, self.name)
+ def __get__(s, obj, owner):
+ if obj is None:
+ return s
+ return getattr(obj, self.name)
+ self.instrument = SynonymProp()
+
+ sessionlib.register_attribute(class_, self.key, uselist=False, proxy_property=self.instrument, useobject=False, comparator=aliased_property.comparator)
+
+ def merge(self, session, source, dest, _recursive):
+ pass
+SynonymProperty.logger = logging.class_logger(SynonymProperty)
+
class PropertyLoader(StrategizedProperty):
"""Describes an object property that holds a single item or list
of items that correspond to a related database table.
return attributes.GenericBackrefExtension(self.key)
mapper.ColumnProperty = ColumnProperty
-
+mapper.SynonymProperty = SynonymProperty
u = s.get(User, 7)
assert u._user_name=='jack'
assert u._user_id ==7
- assert not hasattr(u, 'user_name')
u2 = s.query(User).filter_by(user_name='jack').one()
assert u is u2
))
assert hasattr(User, 'adlist')
- assert not hasattr(User, 'adname')
+ assert hasattr(User, 'adname') # as of 0.4.2, synonyms always create a property
- u = sess.query(User).get_by(uname='jack')
- self.assert_result(u.adlist, Address, *(user_address_result[0]['addresses'][1]))
+ # test compile
+ assert not isinstance(User.uname == 'jack', bool)
- assert hasattr(u, 'adlist')
- assert not hasattr(u, 'adname')
+ u = sess.query(User).filter(User.uname=='jack').one()
+ self.assert_result(u.adlist, Address, *(user_address_result[0]['addresses'][1]))
addr = sess.query(Address).get_by(address_id=user_address_result[0]['addresses'][1][0]['address_id'])
- u = sess.query(User).get_by(adname=addr)
- u2 = sess.query(User).get_by(adlist=addr)
+ u = sess.query(User).filter_by(adname=addr).one()
+ u2 = sess.query(User).filter_by(adlist=addr).one()
+
assert u is u2
assert u not in sess.dirty
assert u.uname == "some user name"
assert u.user_name == "some user name"
assert u in sess.dirty
+
+ def test_column_synonyms(self):
+ """test new-style synonyms which automatically instrument properties, set up aliased column, etc."""
+ sess = create_session()
+
+ assert_col = []
+ class User(object):
+ def _get_user_name(self):
+ assert_col.append(('get', self._user_name))
+ return self._user_name
+ def _set_user_name(self, name):
+ assert_col.append(('set', name))
+ self._user_name = name
+ user_name = property(_get_user_name, _set_user_name)
+
+ mapper(Address, addresses)
+ try:
+ mapper(User, users, properties = {
+ 'addresses':relation(Address, lazy=True),
+ 'not_user_name':synonym('_user_name', map_column=True)
+ })
+ User.not_user_name
+ assert False
+ except exceptions.ArgumentError, e:
+ assert str(e) == "Can't compile synonym '_user_name': no column on table 'users' named 'not_user_name'"
+
+ clear_mappers()
+
+ mapper(Address, addresses)
+ mapper(User, users, properties = {
+ 'addresses':relation(Address, lazy=True),
+ 'user_name':synonym('_user_name', map_column=True)
+ })
+
+ # test compile
+ assert not isinstance(User.user_name == 'jack', bool)
+
+ assert hasattr(User, 'user_name')
+ assert hasattr(User, '_user_name')
+
+ u = sess.query(User).filter(User.user_name == 'jack').one()
+ assert u.user_name == 'jack'
+ u.user_name = 'foo'
+ assert u.user_name == 'foo'
+ assert assert_col == [('get', 'jack'), ('set', 'foo'), ('get', 'foo')]
+
@testing.fails_on('maxdb')
def test_synonymoptions(self):
sess = create_session()
print repr(u.user_id), repr(userlist[0].user_id), repr(userlist[0].user_name)
self.assert_(u.user_id == userlist[0].user_id and userlist[0].user_name == 'modifiedname')
self.assert_(u2.user_id == userlist[1].user_id and userlist[1].user_name == 'savetester2')
-
+
+ def test_synonym(self):
+ class User(object):
+ def _get_name(self):
+ return "User:" + self.user_name
+ def _set_name(self, name):
+ self.user_name = name + ":User"
+ name = property(_get_name, _set_name)
+
+ mapper(User, users, properties={
+ 'name':synonym('user_name')
+ })
+
+ u = User()
+ u.name = "some name"
+ assert u.name == 'User:some name:User'
+ Session.save(u)
+ Session.flush()
+ Session.clear()
+ u = Session.query(User).first()
+ assert u.name == 'User:some name:User'
+
def test_lazyattr_commit(self):
"""tests that when a lazy-loaded list is unloaded, and a commit occurs, that the
'passive' call on that list does not blow away its value"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
