--- /dev/null
+.. change::
+ :tags: usecase, orm
+ :tickets: 11575
+
+ The :paramref:`_orm.aliased.name` parameter to :func:`_orm.aliased` may now
+ be combined with the :paramref:`_orm.aliased.flat` parameter, producing
+ per-table names based on a name-prefixed naming convention. Pull request
+ courtesy Eric Atkin.
supported by all modern databases with regards to right-nested joins
and generally produces more efficient queries.
+ When :paramref:`_orm.aliased.flat` is combined with
+ :paramref:`_orm.aliased.name`, the resulting joins will alias individual
+ tables using a naming scheme similar to ``<prefix>_<tablename>``. This
+ naming scheme is for visibility / debugging purposes only and the
+ specific scheme is subject to change without notice.
+
+ .. versionadded:: 2.0.32 added support for combining
+ :paramref:`_orm.aliased.name` with :paramref:`_orm.aliased.flat`.
+ Previously, this would raise ``NotImplementedError``.
+
:param adapt_on_names: if True, more liberal "matching" will be used when
mapping the mapped columns of the ORM entity to those of the
given selectable - a name-based match will be performed if the
) -> TODO_Any:
sqlutil = util.preloaded.sql_util
if flat:
- if name is not None:
- raise exc.ArgumentError("Can't send name argument with flat")
+ if isinstance(self.left, (FromGrouping, Join)):
+ left_name = name # will recurse
+ else:
+ if name and isinstance(self.left, NamedFromClause):
+ left_name = f"{name}_{self.left.name}"
+ else:
+ left_name = name
+ if isinstance(self.right, (FromGrouping, Join)):
+ right_name = name # will recurse
+ else:
+ if name and isinstance(self.right, NamedFromClause):
+ right_name = f"{name}_{self.right.name}"
+ else:
+ right_name = name
left_a, right_a = (
- self.left._anonymous_fromclause(flat=True),
- self.right._anonymous_fromclause(flat=True),
+ self.left._anonymous_fromclause(name=left_name, flat=flat),
+ self.right._anonymous_fromclause(name=right_name, flat=flat),
)
adapter = sqlutil.ClauseAdapter(left_a).chain(
sqlutil.ClauseAdapter(right_a)
"anon_1.primary_language FROM anon_1",
)
+ @testing.variation("named", [True, False])
+ @testing.variation("flat", [True, False])
+ def test_aliased_joined_entities(self, named, flat):
+ Company = self.classes.Company
+ Engineer = self.classes.Engineer
+
+ if named:
+ e1 = aliased(Engineer, flat=flat, name="myengineer")
+ else:
+ e1 = aliased(Engineer, flat=flat)
+
+ q = select(Company.name, e1.primary_language).join(
+ Company.employees.of_type(e1)
+ )
+
+ if not flat:
+ name = "anon_1" if not named else "myengineer"
+
+ self.assert_compile(
+ q,
+ "SELECT companies.name, "
+ f"{name}.engineers_primary_language FROM companies "
+ "JOIN (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 "
+ "FROM people JOIN engineers "
+ "ON people.person_id = engineers.person_id) AS "
+ f"{name} "
+ f"ON companies.company_id = {name}.people_company_id",
+ )
+ elif named:
+ self.assert_compile(
+ q,
+ "SELECT companies.name, "
+ "myengineer_engineers.primary_language "
+ "FROM companies JOIN (people AS myengineer_people "
+ "JOIN engineers AS myengineer_engineers "
+ "ON myengineer_people.person_id = "
+ "myengineer_engineers.person_id) "
+ "ON companies.company_id = myengineer_people.company_id",
+ )
+ else:
+ self.assert_compile(
+ q,
+ "SELECT companies.name, engineers_1.primary_language "
+ "FROM companies JOIN (people AS people_1 "
+ "JOIN engineers AS engineers_1 "
+ "ON people_1.person_id = engineers_1.person_id) "
+ "ON companies.company_id = people_1.company_id",
+ )
+
class RawSelectTest(QueryTest, AssertsCompiledSQL):
"""older tests from test_query. Here, they are converted to use
"a AS a_1 JOIN b AS b_1 ON a_1.a = b_1.b",
)
+ def test_join_alias_name_flat(self):
+ a = table("a", column("a"))
+ b = table("b", column("b"))
+ self.assert_compile(
+ a.join(b, a.c.a == b.c.b)._anonymous_fromclause(
+ name="foo", flat=True
+ ),
+ "a AS foo_a JOIN b AS foo_b ON foo_a.a = foo_b.b",
+ )
+
def test_composed_join_alias_flat(self):
a = table("a", column("a"))
b = table("b", column("b"))
"ON b_1.b = c_1.c",
)
+ def test_composed_join_alias_name_flat(self):
+ a = table("a", column("a"))
+ b = table("b", column("b"))
+ c = table("c", column("c"))
+ d = table("d", column("d"))
+
+ j1 = a.join(b, a.c.a == b.c.b)
+ j2 = c.join(d, c.c.c == d.c.d)
+
+ self.assert_compile(
+ j1.join(j2, b.c.b == c.c.c)._anonymous_fromclause(
+ name="foo", flat=True
+ ),
+ "a AS foo_a JOIN b AS foo_b ON foo_a.a = foo_b.b JOIN "
+ "(c AS foo_c JOIN d AS foo_d ON foo_c.c = foo_d.d) "
+ "ON foo_b.b = foo_c.c",
+ )
+
def test_composed_join_alias(self):
a = table("a", column("a"))
b = table("b", column("b"))
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
