- fixed endless loop issue when using lazy="dynamic" on both
sides of a bi-directional relationship [ticket:872]
+ - fixed bug where Query would not apply a subquery to the SQL when LIMIT
+ was used in conjunction with an inheriting mapper where the eager
+ loader was only in the parent mapper.
+
- clarified the error message which occurs when you try to update()
an instance with the same identity key as an instance already present
in the session.
raise exceptions.InvalidRequestError("No contextual Session is established. Use a MapperExtension that implements get_session or use 'import sqlalchemy.mods.threadlocal' to establish a default thread-local contextual session.")
- def has_eager(self):
- """Return True if one of the properties attached to this
- Mapper is eager loading.
- """
-
- return len(self._eager_loaders) > 0
-
def instances(self, cursor, session, *mappers, **kwargs):
"""Return a list of mapped instances corresponding to the rows
in a given ResultProxy.
from sqlalchemy import sql, util, exceptions, logging
from sqlalchemy.sql import util as sql_util
-from sqlalchemy.sql import expression, visitors
+from sqlalchemy.sql import expression, visitors, operators
from sqlalchemy.orm import mapper, object_mapper
from sqlalchemy.orm import util as mapperutil
-import operator
+from itertools import chain
__all__ = ['Query', 'QueryContext']
self._populate_existing = False
self._version_check = False
self._autoflush = True
- self._eager_loaders = util.Set([x for x in self.mapper._eager_loaders])
+ self._eager_loaders = util.Set(chain(*[mapper._eager_loaders for mapper in [m for m in self.mapper.iterate_to_root()]]))
self._attributes = {}
self._current_path = ()
self._primary_adapter=None
mapper = object_mapper(instance)
prop = mapper.get_property(property, resolve_synonyms=True)
target = prop.mapper
- criterion = prop.compare(operator.eq, instance, value_is_parent=True)
+ criterion = prop.compare(operators.eq, instance, value_is_parent=True)
return Query(target, **kwargs).filter(criterion)
query_from_parent = classmethod(query_from_parent)
raise exceptions.InvalidRequestError("Could not locate a property which relates instances of class '%s' to instances of class '%s'" % (self.mapper.class_.__name__, instance.__class__.__name__))
else:
prop = mapper.get_property(property, resolve_synonyms=True)
- return self.filter(prop.compare(operator.eq, instance, value_is_parent=True))
+ return self.filter(prop.compare(operators.eq, instance, value_is_parent=True))
def add_entity(self, entity, alias=None, id=None):
"""add a mapped entity to the list of result columns to be returned.
def filter_by(self, **kwargs):
"""apply the given filtering criterion to the query and return the newly resulting ``Query``."""
- clauses = [self._joinpoint.get_property(key, resolve_synonyms=True).compare(operator.eq, value)
+ clauses = [self._joinpoint.get_property(key, resolve_synonyms=True).compare(operators.eq, value)
for key, value in kwargs.iteritems()]
return self.filter(sql.and_(*clauses))
for key, value in params.iteritems():
(keys, prop) = self._locate_prop(key, start=start)
if isinstance(prop, properties.PropertyLoader):
- c = prop.compare(operator.eq, value) & self.join_via(keys[:-1])
+ c = prop.compare(operators.eq, value) & self.join_via(keys[:-1])
else:
- c = prop.compare(operator.eq, value) & self.join_via(keys)
+ c = prop.compare(operators.eq, value) & self.join_via(keys)
if clause is None:
clause = c
else:
# testing a particular endless loop condition in eager join setup
create_session().query(SubT).all()
-
if __name__ == '__main__':
testbase.main()
from sqlalchemy import exceptions, util
from sqlalchemy.orm import *
from testlib import *
-
+from testlib import fixtures
class AttrSettable(object):
def __init__(self, **kwargs):
print class_mapper(Person).primary_key
print person_mapper.get_select_mapper().primary_key
- # so the primaryjoin is "people.c.person_id==cars.c.owner". the "lazy" clause will be
- # "people.c.person_id=?". the employee_join is two selects union'ed together, one of which
- # will contain employee.c.person_id the other contains manager.c.person_id. people.c.person_id is not explicitly in
- # either column clause in this case. we can modify polymorphic_union to always put the "base" column in which would fix this,
- # but im not sure if that really fixes the issue in all cases and its too far from the problem.
- # instead, when the primaryjoin is adapted to point to the polymorphic union and is targeting employee_join.c.person_id,
- # it has to use not just straight column correspondence but also "keys_ok=True", meaning it will link up to any column
- # with the name "person_id", as opposed to columns that descend directly from people.c.person_id. polymorphic unions
- # require the cols all match up on column name which then determine the top selectable names, so matching by name is OK.
-
session = create_session()
# creating 5 managers named from M1 to E5
ot1 = sess.query(T1).get(ot1.id)
ot1.data = 'hi'
sess.flush()
+
+class InheritingEagerTest(ORMTest):
+ def define_tables(self, metadata):
+ global people, employees, tags, peopleTags
+
+ people = Table('people', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('_type', String(30), nullable=False),
+ )
+
+
+ employees = Table('employees', metadata,
+ Column('id', Integer, ForeignKey('people.id'),primary_key=True),
+ )
+
+ tags = Table('tags', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('label', String, nullable=False),
+ )
+
+ peopleTags = Table('peopleTags', metadata,
+ Column('person_id', Integer,ForeignKey('people.id')),
+ Column('tag_id', Integer,ForeignKey('tags.id')),
+ )
+
+ def test_basic(self):
+ """test that Query uses the full set of mapper._eager_loaders when generating SQL"""
+
+ class Person(fixtures.Base):
+ pass
+
+ class Employee(Person):
+ def __init__(self, name='bob'):
+ self.name = name
+
+ class Tag(fixtures.Base):
+ def __init__(self, label):
+ self.label = label
+
+ mapper(Person, people, polymorphic_on=people.c._type,polymorphic_identity='person', properties={
+ 'tags': relation(Tag, secondary=peopleTags,backref='people', lazy=False)
+ })
+ mapper(Employee, employees, inherits=Person,polymorphic_identity='employee')
+ mapper(Tag, tags)
+
+ session = create_session()
+
+ bob = Employee()
+ session.save(bob)
+
+ tag = Tag('crazy')
+ bob.tags.append(tag)
+
+ tag = Tag('funny')
+ bob.tags.append(tag)
+ session.flush()
+
+ session.clear()
+ # query from Employee with limit, query needs to apply eager limiting subquery
+ instance = session.query(Employee).filter_by(id=1).limit(1).first()
+ assert len(instance.tags) == 2
+
if __name__ == "__main__":
testbase.main()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
