when configured on a joined-table subclass, introduced in
version 0.5.6 as a result of the fix for [ticket:1480].
[ticket:1616] thx to Scott Torborg.
+
+ - The "use get" behavior of many-to-one relations, i.e. that a
+ lazy load will fallback to the possibly cached query.get()
+ value, now works across join conditions where the two compared
+ types are not exactly the same class, but share the same
+ "affinity" - i.e. Integer and SmallInteger. Also allows
+ combinations of reflected and non-reflected types to work
+ with 0.5 style type reflection, such as PGText/Text (note 0.6
+ reflects types as their generic versions). [ticket:1556]
- sql
- Fixed bug in two-phase transaction whereby commit() method
(i.e. _CursorFairy) now proxies `__iter__()` to the
underlying cursor correctly. [ticket:1632]
+ - types now support an "affinity comparison" operation, i.e.
+ that an Integer/SmallInteger are "compatible", or
+ a Text/String, PickleType/Binary, etc. Part of
+ [ticket:1556].
+
- sqlite
- sqlite dialect properly generates CREATE INDEX for a table
that is in an alternate schema. [ticket:1439]
# determine if our "lazywhere" clause is the same as the mapper's
# get() clause. then we can just use mapper.get()
- #from sqlalchemy.orm import query
self.use_get = not self.uselist and self.mapper._get_clause[0].compare(self.__lazywhere)
if self.use_get:
self.logger.info("%s will use query.get() to optimize instance loads" % self)
else:
return obj.type
- def compare(self, other):
- """Compare this ``_BindParamClause`` to the given clause.
-
- Since ``compare()`` is meant to compare statement syntax, this
- method returns True if the two ``_BindParamClauses`` have just
- the same type.
-
- """
- return isinstance(other, _BindParamClause) and other.type.__class__ == self.type.__class__ and self.value == other.value
+ def compare(self, other, **kw):
+ """Compare this ``_BindParamClause`` to the given clause."""
+
+ return isinstance(other, _BindParamClause) and \
+ self.type._compare_type_affinity(other.type) and \
+ self.value == other.value
def __getstate__(self):
"""execute a deferred value for serialization purposes."""
By default, returns the operator unchanged.
"""
return op
+
+ @util.memoized_property
+ def _type_affinity(self):
+ """Return a rudimental 'affinity' value expressing the general class of type."""
+
+ for i, t in enumerate(self.__class__.__mro__):
+ if t is TypeEngine:
+ return self.__class__.__mro__[i - 1]
+ else:
+ return self.__class__
+
+ def _compare_type_affinity(self, other):
+ return self._type_affinity is other._type_affinity
def __repr__(self):
return "%s(%s)" % (
self._impl_dict[dialect] = tt
return tt
+ @util.memoized_property
+ def _type_affinity(self):
+ return self.impl._type_affinity
+
def load_dialect_impl(self, dialect):
"""Loads the dialect-specific implementation of this type.
from sqlalchemy.orm import attributes
import sqlalchemy as sa
from sqlalchemy.test import testing
-from sqlalchemy import Integer, String, ForeignKey
+from sqlalchemy import Integer, String, ForeignKey, SmallInteger
+from sqlalchemy.types import TypeDecorator
from sqlalchemy.test.schema import Table
from sqlalchemy.test.schema import Column
from sqlalchemy.orm import mapper, relation, create_session
self.assert_sql_count(testing.db, go, 0)
sa.orm.clear_mappers()
+ @testing.resolve_artifact_names
+ def test_uses_get_compatible_types(self):
+ """test the use_get optimization with compatible but non-identical types"""
+
+ class IntDecorator(TypeDecorator):
+ impl = Integer
+
+ class SmallintDecorator(TypeDecorator):
+ impl = SmallInteger
+
+ class SomeDBInteger(sa.Integer):
+ pass
+
+ for tt in [
+ Integer,
+ SmallInteger,
+ IntDecorator,
+ SmallintDecorator,
+ SomeDBInteger,
+ ]:
+ m = sa.MetaData()
+ users = Table('users', m,
+ Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('name', String(30), nullable=False),
+ )
+ addresses = Table('addresses', m,
+ Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('user_id', tt, ForeignKey('users.id')),
+ Column('email_address', String(50), nullable=False),
+ )
+
+ mapper(Address, addresses, properties = dict(
+ user = relation(mapper(User, users))
+ ))
+
+ sess = create_session(bind=testing.db)
+
+ # load address
+ a1 = sess.query(Address).filter_by(email_address="ed@wood.com").one()
+
+ # load user that is attached to the address
+ u1 = sess.query(User).get(8)
+
+ def go():
+ # lazy load of a1.user should get it from the session
+ assert a1.user is u1
+ self.assert_sql_count(testing.db, go, 0)
+ sa.orm.clear_mappers()
+
@testing.resolve_artifact_names
def test_many_to_one(self):
mapper(Address, addresses, properties = dict(
]:
assert isinstance(start.dialect_impl(dialect), test), "wanted %r got %r" % (test, start.dialect_impl(dialect))
+class TypeAffinityTest(TestBase):
+ def test_type_affinity(self):
+ for t1, t2, comp in [
+ (Integer(), SmallInteger(), True),
+ (Integer(), String(), False),
+ (Integer(), Integer(), True),
+ (Text(), String(), True),
+ (Text(), Unicode(), True),
+ (Binary(), Integer(), False),
+ (Binary(), PickleType(), True),
+ (PickleType(), Binary(), True),
+ (PickleType(), PickleType(), True),
+ ]:
+ eq_(t1._compare_type_affinity(t2), comp, "%s %s" % (t1, t2))
+
class UserDefinedTest(TestBase):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
