and the AliasedClass if any is referenced.
"""
- def __init__(self, entity, equivalents=None,
- chain_to=None, adapt_required=False):
+ def __init__(self, entity, equivalents=None, adapt_required=False,
+ chain_to=None):
info = inspection.inspect(entity)
self.mapper = info.mapper
"""
def __init__(self, cls, alias=None,
name=None,
- flat=True,
+ flat=False,
adapt_on_names=False,
# TODO: None for default here?
with_polymorphic_mappers=(),
use_mapper_path=False):
mapper = _class_to_mapper(cls)
if alias is None:
- alias = mapper._with_polymorphic_selectable.alias(name=name, flat=flat)
+ alias = mapper._with_polymorphic_selectable.alias(
+ name=name, flat=flat)
self._aliased_insp = AliasedInsp(
self,
mapper,
inspection._inspects(AliasedInsp)(lambda target: target)
-def aliased(element, alias=None, name=None, adapt_on_names=False):
+def aliased(element, alias=None, name=None, flat=False, adapt_on_names=False):
"""Produce an alias of the given element, usually an :class:`.AliasedClass`
instance.
raise sa_exc.ArgumentError(
"adapt_on_names only applies to ORM elements"
)
- return element.alias(name)
+ return element.alias(name, flat=flat)
else:
- return AliasedClass(element, alias=alias,
+ return AliasedClass(element, alias=alias, flat=flat,
name=name, adapt_on_names=adapt_on_names)
def with_polymorphic(base, classes, selectable=False,
+ flat=False,
polymorphic_on=None, aliased=False,
innerjoin=False, _use_mapper_path=False):
"""Produce an :class:`.AliasedClass` construct which specifies
mappers, selectable = primary_mapper.\
_with_polymorphic_args(classes, selectable,
innerjoin=innerjoin)
- if aliased:
- selectable = selectable.alias(flat=True)
+ if aliased or flat:
+ selectable = selectable.alias(flat=flat)
return AliasedClass(base,
selectable,
with_polymorphic_mappers=mappers,
__dialect__ = 'default'
def test_aliased_not_polluted_by_join(self):
+ # aliased(polymorphic) will normally do the old-school
+ # "(SELECT * FROM a JOIN b ...) AS anon_1" thing.
+ # this is the safest
sess = create_session()
palias = aliased(Person)
self.assert_compile(
sess.query(palias, Company.name)
.join(Person, Company.employees)
.filter(palias.name == 'dilbert'),
+ "SELECT anon_1.people_person_id AS anon_1_people_person_id, "
+ "anon_1.people_company_id AS anon_1_people_company_id, "
+ "anon_1.people_name AS anon_1_people_name, "
+ "anon_1.people_type AS anon_1_people_type, "
+ "anon_1.engineers_person_id AS anon_1_engineers_person_id, "
+ "anon_1.engineers_status AS anon_1_engineers_status, "
+ "anon_1.engineers_engineer_name AS anon_1_engineers_engineer_name, "
+ "anon_1.engineers_primary_language AS "
+ "anon_1_engineers_primary_language, "
+ "anon_1.managers_person_id AS anon_1_managers_person_id, "
+ "anon_1.managers_status AS anon_1_managers_status, "
+ "anon_1.managers_manager_name AS anon_1_managers_manager_name, "
+ "anon_1.boss_boss_id AS anon_1_boss_boss_id, "
+ "anon_1.boss_golf_swing AS anon_1_boss_golf_swing, "
+ "companies.name AS companies_name "
+ "FROM (SELECT people.person_id AS people_person_id, "
+ "people.company_id AS people_company_id, "
+ "people.name AS people_name, people.type AS people_type, "
+ "engineers.person_id AS engineers_person_id, "
+ "engineers.status AS engineers_status, "
+ "engineers.engineer_name AS engineers_engineer_name, "
+ "engineers.primary_language AS engineers_primary_language, "
+ "managers.person_id AS managers_person_id, "
+ "managers.status AS managers_status, "
+ "managers.manager_name AS managers_manager_name, "
+ "boss.boss_id AS boss_boss_id, "
+ "boss.golf_swing AS boss_golf_swing "
+ "FROM people LEFT OUTER JOIN engineers "
+ "ON people.person_id = engineers.person_id "
+ "LEFT OUTER JOIN managers "
+ "ON people.person_id = managers.person_id LEFT OUTER JOIN boss "
+ "ON managers.person_id = boss.boss_id) AS anon_1, "
+ "companies JOIN "
+ "(people LEFT OUTER JOIN engineers "
+ "ON people.person_id = engineers.person_id "
+ "LEFT OUTER JOIN managers "
+ "ON people.person_id = managers.person_id "
+ "LEFT OUTER JOIN boss ON managers.person_id = boss.boss_id) "
+ "ON companies.company_id = people.company_id "
+ "WHERE anon_1.people_name = :people_name_1 "
+ "ORDER BY anon_1.people_person_id"
+ )
+
+ def test_flat_aliased_w_select_from(self):
+ sess = create_session()
+ palias = aliased(Person, flat=True)
+ self.assert_compile(
+ sess.query(palias, Company.name)
+ .select_from(palias)
+ .join(Person, Company.employees)
+ .filter(palias.name == 'dilbert'),
"SELECT people_1.person_id AS people_1_person_id, "
"people_1.company_id AS people_1_company_id, "
"people_1.name AS people_1_name, people_1.type AS people_1_type, "
"WHERE people_1.name = :name_1 ORDER BY people_1.person_id"
)
- def test_mixed_entities_compiled_four(self):
- sess = create_session()
- palias = aliased(Person)
- self.assert_compile(
- sess.query(palias, Company.name, Person)
- .join(Company.employees)
- .filter(Company.name == 'Elbonia, Inc.')
- .filter(palias.name == 'dilbert'),
- ""
- )
class PolymorphicUnionsTest(_PolymorphicTestBase, _PolymorphicUnions):
pass
contains_eager, joinedload, subqueryload, relationship,\
subqueryload_all, joinedload_all
from sqlalchemy import and_
-from sqlalchemy import testing
+from sqlalchemy import testing, exc as sa_exc
from sqlalchemy.testing import fixtures
-from sqlalchemy.testing import assert_raises, eq_
+from sqlalchemy.testing import assert_raises, assert_raises_message, eq_
from sqlalchemy.testing.schema import Column
from sqlalchemy.engine import default
from sqlalchemy.testing.entities import ComparableEntity
def test_with_polymorphic_join_compile_one(self):
sess = Session()
-# MARKMARK
+
self.assert_compile(
sess.query(Company).join(
Company.employees.of_type(
- with_polymorphic(Person, [Engineer, Manager], aliased=True)
+ with_polymorphic(Person, [Engineer, Manager],
+ aliased=True, flat=True)
)
),
"SELECT companies.company_id AS companies_company_id, "
def test_with_polymorphic_join_exec_contains_eager_one(self):
sess = Session()
def go():
- wp = with_polymorphic(Person, [Engineer, Manager], aliased=True)
+ wp = with_polymorphic(Person, [Engineer, Manager],
+ aliased=True, flat=True)
eq_(
sess.query(Company).join(
Company.employees.of_type(wp)
def test_subqueryload_explicit_withpoly(self):
sess = Session()
def go():
- target = with_polymorphic(Person, Engineer, aliased=True)
+ target = with_polymorphic(Person, Engineer)
eq_(
sess.query(Company).\
filter_by(company_id=1).\
def test_joinedload_explicit_withpoly(self):
sess = Session()
def go():
- target = with_polymorphic(Person, Engineer, aliased=True)
+ target = with_polymorphic(Person, Engineer, flat=True)
eq_(
sess.query(Company).\
filter_by(company_id=1).\
comp_sel.process(sel, asfrom=True).replace("\n", "") + \
" ON companies.company_id = people_1.company_id"
+ def test_joinedload_explicit_with_unaliased_poly_compile(self):
+ sess = Session()
+ target = with_polymorphic(Person, Engineer)
+ q = sess.query(Company).\
+ filter_by(company_id=1).\
+ options(joinedload(Company.employees.of_type(target)))
+ assert_raises_message(
+ sa_exc.InvalidRequestError,
+ "Detected unaliased columns when generating joined load.",
+ q._compile_context
+ )
+
+
+ def test_joinedload_explicit_with_flataliased_poly_compile(self):
+ sess = Session()
+ target = with_polymorphic(Person, Engineer, flat=True)
+ q = sess.query(Company).\
+ filter_by(company_id=1).\
+ options(joinedload(Company.employees.of_type(target)))
+ self.assert_compile(q,
+ "SELECT companies.company_id AS companies_company_id, "
+ "companies.name AS companies_name, "
+ "people_1.person_id AS people_1_person_id, "
+ "people_1.company_id AS people_1_company_id, "
+ "people_1.name AS people_1_name, people_1.type AS people_1_type, "
+ "engineers_1.person_id AS engineers_1_person_id, "
+ "engineers_1.status AS engineers_1_status, "
+ "engineers_1.engineer_name AS engineers_1_engineer_name, "
+ "engineers_1.primary_language AS engineers_1_primary_language "
+ "FROM companies LEFT OUTER JOIN (people AS people_1 "
+ "LEFT OUTER JOIN engineers AS engineers_1 "
+ "ON people_1.person_id = engineers_1.person_id "
+ "LEFT OUTER JOIN managers AS managers_1 "
+ "ON people_1.person_id = managers_1.person_id) "
+ "ON companies.company_id = people_1.company_id "
+ "WHERE companies.company_id = :company_id_1 "
+ "ORDER BY people_1.person_id"
+ )
class SubclassRelationshipTest(testing.AssertsCompiledSQL, fixtures.DeclarativeMappedTest):
"""There's overlap here vs. the ones above."""
self.classes.Job,\
self.classes.SubJob
- Job_P = with_polymorphic(Job, SubJob, aliased=True)
+ Job_P = with_polymorphic(Job, SubJob, aliased=True, flat=True)
s = Session()
q = s.query(Job).join(DataContainer.jobs).\
self.classes.Job,\
self.classes.SubJob
- Job_P = with_polymorphic(Job, SubJob, aliased=True)
+ Job_P = with_polymorphic(Job, SubJob)
s = Session()
q = s.query(DataContainer).join(DataContainer.jobs.of_type(Job_P))
"SELECT data_container.id AS data_container_id, "
"data_container.name AS data_container_name "
"FROM data_container JOIN "
- "(job AS job_1 LEFT OUTER JOIN subjob AS subjob_1 "
- "ON job_1.id = subjob_1.id) "
- "ON data_container.id = job_1.container_id")
+ "(job LEFT OUTER JOIN subjob "
+ "ON job.id = subjob.id) "
+ "ON data_container.id = job.container_id")
def test_join_wsubclass(self):
ParentThing, DataContainer, Job, SubJob = \
self.classes.Job,\
self.classes.SubJob
- Job_P = with_polymorphic(Job, SubJob, aliased=True, innerjoin=True)
+ Job_P = with_polymorphic(Job, SubJob, innerjoin=True)
s = Session()
q = s.query(DataContainer).join(DataContainer.jobs.of_type(Job_P))
"SELECT data_container.id AS data_container_id, "
"data_container.name AS data_container_name "
"FROM data_container JOIN "
- "(job AS job_1 JOIN subjob AS subjob_1 ON job_1.id = subjob_1.id) "
- "ON data_container.id = job_1.container_id")
+ "(job JOIN subjob ON job.id = subjob.id) "
+ "ON data_container.id = job.container_id")
def test_join_walias(self):
ParentThing, DataContainer, Job, SubJob = \
"FROM data_container JOIN job AS job_1 "
"ON data_container.id = job_1.container_id")
- def test_join_explicit_wpoly(self):
+ def test_join_explicit_wpoly_noalias(self):
ParentThing, DataContainer, Job, SubJob = \
self.classes.ParentThing,\
self.classes.DataContainer,\
self.classes.Job,\
self.classes.SubJob
- Job_P = with_polymorphic(Job, SubJob, aliased=True)
+ Job_P = with_polymorphic(Job, SubJob)
+
+ s = Session()
+ q = s.query(DataContainer).join(Job_P, DataContainer.jobs)
+ self.assert_compile(q,
+ "SELECT data_container.id AS data_container_id, "
+ "data_container.name AS data_container_name "
+ "FROM data_container JOIN "
+ "(job LEFT OUTER JOIN subjob "
+ "ON job.id = subjob.id) "
+ "ON data_container.id = job.container_id")
+
+
+ def test_join_explicit_wpoly_flat(self):
+ ParentThing, DataContainer, Job, SubJob = \
+ self.classes.ParentThing,\
+ self.classes.DataContainer,\
+ self.classes.Job,\
+ self.classes.SubJob
+
+ Job_P = with_polymorphic(Job, SubJob, flat=True)
s = Session()
q = s.query(DataContainer).join(Job_P, DataContainer.jobs)
"ON job_1.id = subjob_1.id) "
"ON data_container.id = job_1.container_id")
+ def test_join_explicit_wpoly_full_alias(self):
+ ParentThing, DataContainer, Job, SubJob = \
+ self.classes.ParentThing,\
+ self.classes.DataContainer,\
+ self.classes.Job,\
+ self.classes.SubJob
+
+ Job_P = with_polymorphic(Job, SubJob, aliased=True)
+
+ s = Session()
+ q = s.query(DataContainer).join(Job_P, DataContainer.jobs)
+ self.assert_compile(q,
+ "SELECT data_container.id AS data_container_id, "
+ "data_container.name AS data_container_name "
+ "FROM data_container JOIN "
+ "(SELECT job.id AS job_id, job.type AS job_type, "
+ "job.container_id AS job_container_id, "
+ "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
+ "FROM job LEFT OUTER JOIN subjob ON job.id = subjob.id) "
+ "AS anon_1 ON data_container.id = anon_1.job_container_id"
+ )
+
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
