event.listen(desc, 'append', append, raw=True, retval=True)
event.listen(desc, 'set', set_, raw=True, retval=True)
-def polymorphic_union(table_map, typecolname, aliasname='p_union'):
+def polymorphic_union(table_map, typecolname, aliasname='p_union', cast_nulls=True):
"""Create a ``UNION`` statement used by a polymorphic mapper.
See :ref:`concrete_inheritance` for an example of how
this is used.
+
+ :param table_map: mapping of polymorphic identities to
+ :class:`.Table` objects.
+ :param typecolname: string name of a "discriminator" column, which will be
+ derived from the query, producing the polymorphic identity for each row. If
+ ``None``, no polymorphic discriminator is generated.
+ :param aliasname: name of the :func:`~sqlalchemy.sql.expression.alias()`
+ construct generated.
+ :param cast_nulls: if True, non-existent columns, which are represented as labeled
+ NULLs, will be passed into CAST. This is a legacy behavior that is problematic
+ on some backends such as Oracle - in which case it can be set to False.
+
"""
- colnames = set()
+ colnames = util.OrderedSet()
colnamemaps = {}
types = {}
for key in table_map.keys():
try:
return colnamemaps[table][name]
except KeyError:
- return sql.cast(sql.null(), types[name]).label(name)
+ if cast_nulls:
+ return sql.cast(sql.null(), types[name]).label(name)
+ else:
+ return sql.type_coerce(sql.null(), types[name]).label(name)
result = []
for type, table in table_map.iteritems():
from sqlalchemy.orm import *
from sqlalchemy.orm import exc as orm_exc, attributes
from test.lib.assertsql import AllOf, CompiledSQL
-
+from sqlalchemy.sql import table, column
from test.lib import testing, engines
from test.lib import fixtures
from test.orm import _fixtures
sess.add(s1)
assert_raises(sa_exc.DBAPIError, sess.flush)
+class PolymorphicUnionTest(fixtures.TestBase, testing.AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def _fixture(self):
+ t1 = table('t1', column('c1', Integer),
+ column('c2', Integer),
+ column('c3', Integer))
+ t2 = table('t2', column('c1', Integer), column('c2', Integer),
+ column('c3', Integer),
+ column('c4', Integer))
+ t3 = table('t3', column('c1', Integer),
+ column('c3', Integer),
+ column('c5', Integer))
+ return t1, t2, t3
+
+ def test_type_col_present(self):
+ t1, t2, t3 = self._fixture()
+ self.assert_compile(
+ polymorphic_union(
+ util.OrderedDict([("a", t1), ("b", t2), ("c", t3)]),
+ 'q1'
+ ),
+ "SELECT t1.c1, t1.c2, t1.c3, CAST(NULL AS INTEGER) AS c4, "
+ "CAST(NULL AS INTEGER) AS c5, 'a' AS q1 FROM t1 UNION ALL "
+ "SELECT t2.c1, t2.c2, t2.c3, t2.c4, CAST(NULL AS INTEGER) AS c5, "
+ "'b' AS q1 FROM t2 UNION ALL SELECT t3.c1, "
+ "CAST(NULL AS INTEGER) AS c2, t3.c3, CAST(NULL AS INTEGER) AS c4, "
+ "t3.c5, 'c' AS q1 FROM t3"
+ )
+
+ def test_type_col_non_present(self):
+ t1, t2, t3 = self._fixture()
+ self.assert_compile(
+ polymorphic_union(
+ util.OrderedDict([("a", t1), ("b", t2), ("c", t3)]),
+ None
+ ),
+ "SELECT t1.c1, t1.c2, t1.c3, CAST(NULL AS INTEGER) AS c4, "
+ "CAST(NULL AS INTEGER) AS c5 FROM t1 UNION ALL SELECT t2.c1, "
+ "t2.c2, t2.c3, t2.c4, CAST(NULL AS INTEGER) AS c5 FROM t2 "
+ "UNION ALL SELECT t3.c1, CAST(NULL AS INTEGER) AS c2, t3.c3, "
+ "CAST(NULL AS INTEGER) AS c4, t3.c5 FROM t3"
+ )
+
+ def test_no_cast_null(self):
+ t1, t2, t3 = self._fixture()
+ self.assert_compile(
+ polymorphic_union(
+ util.OrderedDict([("a", t1), ("b", t2), ("c", t3)]),
+ 'q1', cast_nulls=False
+ ),
+ "SELECT t1.c1, t1.c2, t1.c3, NULL AS c4, NULL AS c5, 'a' AS q1 "
+ "FROM t1 UNION ALL SELECT t2.c1, t2.c2, t2.c3, t2.c4, NULL AS c5, "
+ "'b' AS q1 FROM t2 UNION ALL SELECT t3.c1, NULL AS c2, t3.c3, "
+ "NULL AS c4, t3.c5, 'c' AS q1 FROM t3"
+ )