- improved support for complex queries embedded into "where" criterion
for query.select() [ticket:449]
- contains_eager('foo') automatically implies eagerload('foo')
+ - mapper options like eagerload(), lazyload(), deferred(), will work for "synonym()"
+ relationships [ticket:485]
- fixed bug where cascade operations incorrectly included deleted collection
items in the cascade [ticket:445]
- fixed relationship deletion error when one-to-many child item is moved to a new
from sqlalchemy.orm import mapper as mapperlib
from sqlalchemy.orm.query import Query
from sqlalchemy.orm.util import polymorphic_union
-from sqlalchemy.orm import properties, strategies
+from sqlalchemy.orm import properties, strategies, interfaces
from sqlalchemy.orm.session import Session as create_session
from sqlalchemy.orm.session import object_session, attribute_manager
"""set up 'name' as a synonym to another MapperProperty.
Used with the 'properties' dictionary sent to mapper()."""
- return properties.SynonymProperty(name, proxy=proxy)
+ return interfaces.SynonymProperty(name, proxy=proxy)
def compile_mappers():
"""compile all mappers that have been defined.
"""returns a compare operation for the columns represented by this MapperProperty to the given value,
which may be a column value or an instance."""
raise NotImplementedError()
-
+
+class SynonymProperty(MapperProperty):
+ def __init__(self, name, proxy=False):
+ self.name = name
+ self.proxy = proxy
+ def setup(self, querycontext, **kwargs):
+ pass
+ def execute(self, selectcontext, instance, row, identitykey, isnew):
+ pass
+ 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 StrategizedProperty(MapperProperty):
"""a MapperProperty which uses selectable strategies to affect loading behavior.
There is a single default strategy selected, and alternate strategies can be selected
mapper = context.mapper
for token in self.key.split('.'):
prop = mapper.props[token]
+ if isinstance(prop, SynonymProperty):
+ prop = mapper.props[prop.name]
mapper = getattr(prop, 'mapper', None)
self.__prop = prop
return prop
from sqlalchemy.orm import util as mapperutil
import sets, random
from sqlalchemy.orm.interfaces import *
-
-
-class SynonymProperty(MapperProperty):
- def __init__(self, name, proxy=False):
- self.name = name
- self.proxy = proxy
- def setup(self, querycontext, **kwargs):
- pass
- def execute(self, selectcontext, instance, row, identitykey, isnew):
- pass
- 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 ColumnProperty(StrategizedProperty):
"""describes an object attribute that corresponds to a table column."""
from sqlalchemy import sql, util, exceptions, sql_util, logging, schema
from sqlalchemy.orm import mapper, class_mapper
-from sqlalchemy.orm.interfaces import OperationContext
+from sqlalchemy.orm.interfaces import OperationContext, SynonymProperty
__all__ = ['Query', 'QueryContext', 'SelectionContext']
seen.add(mapper_)
if mapper_.props.has_key(key):
prop = mapper_.props[key]
- if isinstance(prop, properties.SynonymProperty):
+ if isinstance(prop, SynonymProperty):
prop = mapper_.props[prop.name]
if isinstance(prop, properties.PropertyLoader):
keys.insert(0, prop.key)
assert u.uname == "some user name"
assert u.user_name == "some user name"
assert u in sess.dirty
-
+
+ def testsynonymoptions(self):
+ sess = create_session()
+ mapper(User, users, properties = dict(
+ addresses = relation(mapper(Address, addresses), lazy = True),
+ adlist = synonym('addresses', proxy=True)
+ ))
+
+ def go():
+ u = sess.query(User).options(eagerload('adlist')).get_by(user_name='jack')
+ self.assert_result(u.adlist, Address, *(user_address_result[0]['addresses'][1]))
+ self.assert_sql_count(db, go, 1)
+
def testextensionoptions(self):
sess = create_session()
class ext1(MapperExtension):