from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import literal
from sqlalchemy import null
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy import union
from sqlalchemy import union_all
from sqlalchemy.ext.declarative import AbstractConcreteBase
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import aliased
from sqlalchemy.orm import attributes
from sqlalchemy.orm import class_mapper
from sqlalchemy.orm import clear_mappers
from sqlalchemy.orm import composite
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import contains_eager
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import polymorphic_union
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm import with_polymorphic
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.entities import ComparableEntity
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.fixtures import RemoveORMEventsGlobally
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
class ConcreteTest(AssertsCompiledSQL, fixtures.MappedTest):
__dialect__ = "default"
@classmethod
def define_tables(cls, metadata):
Table(
"companies",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(50)),
)
Table(
"employees",
metadata,
Column(
"employee_id",
Integer,
primary_key=True,
test_needs_autoincrement=True,
),
Column("name", String(50)),
Column("company_id", Integer, ForeignKey("companies.id")),
)
Table(
"managers",
metadata,
Column(
"employee_id",
Integer,
primary_key=True,
test_needs_autoincrement=True,
),
Column("name", String(50)),
Column("manager_data", String(50)),
Column("company_id", Integer, ForeignKey("companies.id")),
)
Table(
"engineers",
metadata,
Column(
"employee_id",
Integer,
primary_key=True,
test_needs_autoincrement=True,
),
Column("name", String(50)),
Column("engineer_info", String(50)),
Column("company_id", Integer, ForeignKey("companies.id")),
)
Table(
"hackers",
metadata,
Column(
"employee_id",
Integer,
primary_key=True,
test_needs_autoincrement=True,
),
Column("name", String(50)),
Column("engineer_info", String(50)),
Column("company_id", Integer, ForeignKey("companies.id")),
Column("nickname", String(50)),
)
@classmethod
def setup_classes(cls):
class Employee(cls.Basic):
def __init__(self, name):
self.name = name
def __repr__(self):
return self.__class__.__name__ + " " + self.name
class Manager(Employee):
def __init__(self, name, manager_data):
self.name = name
self.manager_data = manager_data
def __repr__(self):
return (
self.__class__.__name__
+ " "
+ self.name
+ " "
+ self.manager_data
)
class Engineer(Employee):
def __init__(self, name, engineer_info):
self.name = name
self.engineer_info = engineer_info
def __repr__(self):
return (
self.__class__.__name__
+ " "
+ self.name
+ " "
+ self.engineer_info
)
class Hacker(Engineer):
def __init__(self, name, nickname, engineer_info):
self.name = name
self.nickname = nickname
self.engineer_info = engineer_info
def __repr__(self):
return (
self.__class__.__name__
+ " "
+ self.name
+ " '"
+ self.nickname
+ "' "
+ self.engineer_info
)
class Company(cls.Basic):
pass
def test_basic(self):
Employee, Engineer, Manager = self.classes(
"Employee", "Engineer", "Manager"
)
engineers_table, managers_table = self.tables("engineers", "managers")
pjoin = polymorphic_union(
{"manager": managers_table, "engineer": engineers_table},
"type",
"pjoin",
)
employee_mapper = self.mapper_registry.map_imperatively(
Employee, pjoin, polymorphic_on=pjoin.c.type
)
self.mapper_registry.map_imperatively(
Manager,
managers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="manager",
)
self.mapper_registry.map_imperatively(
Engineer,
engineers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="engineer",
)
session = fixture_session()
session.add(Manager("Sally", "knows how to manage things"))
session.add(Engineer("Karina", "knows how to hack"))
session.flush()
session.expunge_all()
assert {repr(x) for x in session.query(Employee)} == {
"Engineer Karina knows how to hack",
"Manager Sally knows how to manage things",
}
assert {repr(x) for x in session.query(Manager)} == {
"Manager Sally knows how to manage things"
}
assert {repr(x) for x in session.query(Engineer)} == {
"Engineer Karina knows how to hack"
}
manager = session.query(Manager).one()
session.expire(manager, ["manager_data"])
eq_(manager.manager_data, "knows how to manage things")
def test_multi_level_no_base(self):
Employee, Engineer, Manager = self.classes(
"Employee", "Engineer", "Manager"
)
(Hacker,) = self.classes("Hacker")
engineers_table, managers_table = self.tables("engineers", "managers")
(hackers_table,) = self.tables("hackers")
pjoin = polymorphic_union(
{
"manager": managers_table,
"engineer": engineers_table,
"hacker": hackers_table,
},
"type",
"pjoin",
)
pjoin2 = polymorphic_union(
{"engineer": engineers_table, "hacker": hackers_table},
"type",
"pjoin2",
)
employee_mapper = self.mapper_registry.map_imperatively(
Employee, pjoin, polymorphic_on=pjoin.c.type
)
self.mapper_registry.map_imperatively(
Manager,
managers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="manager",
)
engineer_mapper = self.mapper_registry.map_imperatively(
Engineer,
engineers_table,
with_polymorphic=("*", pjoin2),
polymorphic_on=pjoin2.c.type,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="engineer",
)
self.mapper_registry.map_imperatively(
Hacker,
hackers_table,
inherits=engineer_mapper,
concrete=True,
polymorphic_identity="hacker",
)
session = fixture_session()
sally = Manager("Sally", "knows how to manage things")
assert_raises_message(
AttributeError,
"does not implement attribute .?'type' at the instance level.",
setattr,
sally,
"type",
"sometype",
)
self.assert_compile(Manager.type == "x", "pjoin.type = :type_1")
jenn = Engineer("Jenn", "knows how to program")
hacker = Hacker("Karina", "Badass", "knows how to hack")
assert_raises_message(
AttributeError,
"does not implement attribute .?'type' at the instance level.",
setattr,
hacker,
"type",
"sometype",
)
session.add_all((sally, jenn, hacker))
session.flush()
assert (
"nickname"
not in attributes.instance_state(jenn).expired_attributes
)
assert "name" not in attributes.instance_state(jenn).expired_attributes
assert (
"name" not in attributes.instance_state(hacker).expired_attributes
)
assert (
"nickname"
not in attributes.instance_state(hacker).expired_attributes
)
def go():
eq_(jenn.name, "Jenn")
eq_(hacker.nickname, "Badass")
self.assert_sql_count(testing.db, go, 0)
session.expunge_all()
assert (
repr(
session.query(Employee).filter(Employee.name == "Sally").one()
)
== "Manager Sally knows how to manage things"
)
assert (
repr(session.query(Manager).filter(Manager.name == "Sally").one())
== "Manager Sally knows how to manage things"
)
assert {repr(x) for x in session.query(Employee).all()} == {
"Engineer Jenn knows how to program",
"Manager Sally knows how to manage things",
"Hacker Karina 'Badass' knows how to hack",
}
assert {repr(x) for x in session.query(Manager).all()} == {
"Manager Sally knows how to manage things"
}
assert {repr(x) for x in session.query(Engineer).all()} == {
"Engineer Jenn knows how to program",
"Hacker Karina 'Badass' knows how to hack",
}
assert {repr(x) for x in session.query(Hacker).all()} == {
"Hacker Karina 'Badass' knows how to hack"
}
def test_multi_level_no_base_w_hybrid(self):
Employee, Engineer, Manager = self.classes(
"Employee", "Engineer", "Manager"
)
(Hacker,) = self.classes("Hacker")
engineers_table, managers_table = self.tables("engineers", "managers")
(hackers_table,) = self.tables("hackers")
pjoin = polymorphic_union(
{
"manager": managers_table,
"engineer": engineers_table,
"hacker": hackers_table,
},
"type",
"pjoin",
)
test_calls = mock.Mock()
class ManagerWHybrid(Employee):
def __init__(self, name, manager_data):
self.name = name
self.manager_data = manager_data
@hybrid_property
def engineer_info(self):
test_calls.engineer_info_instance()
return self.manager_data
@engineer_info.expression
def engineer_info(cls):
test_calls.engineer_info_class()
return cls.manager_data
employee_mapper = self.mapper_registry.map_imperatively(
Employee, pjoin, polymorphic_on=pjoin.c.type
)
self.mapper_registry.map_imperatively(
ManagerWHybrid,
managers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="manager",
)
self.mapper_registry.map_imperatively(
Engineer,
engineers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="engineer",
)
session = fixture_session()
sally = ManagerWHybrid("Sally", "mgrdata")
eq_(test_calls.mock_calls, [])
eq_(sally.engineer_info, "mgrdata")
eq_(test_calls.mock_calls, [mock.call.engineer_info_instance()])
session.add(sally)
session.commit()
session.close()
Sally = (
session.query(ManagerWHybrid)
.filter(ManagerWHybrid.engineer_info == "mgrdata")
.one()
)
eq_(
test_calls.mock_calls,
[
mock.call.engineer_info_instance(),
mock.call.engineer_info_class(),
],
)
eq_(Sally.engineer_info, "mgrdata")
def test_multi_level_with_base(self):
Employee, Engineer, Manager = self.classes(
"Employee", "Engineer", "Manager"
)
employees_table, engineers_table, managers_table = self.tables(
"employees", "engineers", "managers"
)
(hackers_table,) = self.tables("hackers")
(Hacker,) = self.classes("Hacker")
pjoin = polymorphic_union(
{
"employee": employees_table,
"manager": managers_table,
"engineer": engineers_table,
"hacker": hackers_table,
},
"type",
"pjoin",
)
pjoin2 = polymorphic_union(
{"engineer": engineers_table, "hacker": hackers_table},
"type",
"pjoin2",
)
employee_mapper = self.mapper_registry.map_imperatively(
Employee,
employees_table,
with_polymorphic=("*", pjoin),
polymorphic_on=pjoin.c.type,
)
self.mapper_registry.map_imperatively(
Manager,
managers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="manager",
)
engineer_mapper = self.mapper_registry.map_imperatively(
Engineer,
engineers_table,
with_polymorphic=("*", pjoin2),
polymorphic_on=pjoin2.c.type,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="engineer",
)
self.mapper_registry.map_imperatively(
Hacker,
hackers_table,
inherits=engineer_mapper,
concrete=True,
polymorphic_identity="hacker",
)
session = fixture_session()
sally = Manager("Sally", "knows how to manage things")
jenn = Engineer("Jenn", "knows how to program")
hacker = Hacker("Karina", "Badass", "knows how to hack")
session.add_all((sally, jenn, hacker))
session.flush()
def go():
eq_(jenn.name, "Jenn")
eq_(hacker.nickname, "Badass")
self.assert_sql_count(testing.db, go, 0)
session.expunge_all()
assert (
len(
session.connection()
.execute(session.query(Employee).statement)
.fetchall()
)
== 3
)
assert {repr(x) for x in session.query(Employee)} == {
"Engineer Jenn knows how to program",
"Manager Sally knows how to manage things",
"Hacker Karina 'Badass' knows how to hack",
}
assert {repr(x) for x in session.query(Manager)} == {
"Manager Sally knows how to manage things"
}
assert {repr(x) for x in session.query(Engineer)} == {
"Engineer Jenn knows how to program",
"Hacker Karina 'Badass' knows how to hack",
}
assert {repr(x) for x in session.query(Hacker)} == {
"Hacker Karina 'Badass' knows how to hack"
}
@testing.fixture
def two_pjoin_fixture(self):
Employee, Engineer, Manager = self.classes(
"Employee", "Engineer", "Manager"
)
(Hacker,) = self.classes("Hacker")
(employees_table,) = self.tables("employees")
engineers_table, managers_table = self.tables("engineers", "managers")
(hackers_table,) = self.tables("hackers")
pjoin = polymorphic_union(
{
"employee": employees_table,
"manager": managers_table,
"engineer": engineers_table,
"hacker": hackers_table,
},
"type",
"pjoin",
)
pjoin2 = polymorphic_union(
{"engineer": engineers_table, "hacker": hackers_table},
"type",
"pjoin2",
)
employee_mapper = self.mapper_registry.map_imperatively(
Employee, employees_table, polymorphic_identity="employee"
)
self.mapper_registry.map_imperatively(
Manager,
managers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="manager",
)
engineer_mapper = self.mapper_registry.map_imperatively(
Engineer,
engineers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="engineer",
)
self.mapper_registry.map_imperatively(
Hacker,
hackers_table,
inherits=engineer_mapper,
concrete=True,
polymorphic_identity="hacker",
)
session = fixture_session(expire_on_commit=False)
jdoe = Employee("Jdoe")
sally = Manager("Sally", "knows how to manage things")
jenn = Engineer("Jenn", "knows how to program")
hacker = Hacker("Karina", "Badass", "knows how to hack")
session.add_all((jdoe, sally, jenn, hacker))
session.commit()
return (
session,
Employee,
Engineer,
Manager,
Hacker,
pjoin,
pjoin2,
jdoe,
sally,
jenn,
hacker,
)
def test_without_default_polymorphic_one(self, two_pjoin_fixture):
(
session,
Employee,
Engineer,
Manager,
Hacker,
pjoin,
pjoin2,
jdoe,
sally,
jenn,
hacker,
) = two_pjoin_fixture
wp = with_polymorphic(
Employee, "*", pjoin, polymorphic_on=pjoin.c.type
)
eq_(
sorted([repr(x) for x in session.query(wp)]),
[
"Employee Jdoe",
"Engineer Jenn knows how to program",
"Hacker Karina 'Badass' knows how to hack",
"Manager Sally knows how to manage things",
],
)
eq_(session.get(Employee, jdoe.employee_id), jdoe)
eq_(session.get(Engineer, jenn.employee_id), jenn)
def test_without_default_polymorphic_two(self, two_pjoin_fixture):
(
session,
Employee,
Engineer,
Manager,
Hacker,
pjoin,
pjoin2,
jdoe,
sally,
jenn,
hacker,
) = two_pjoin_fixture
wp = with_polymorphic(
Employee, "*", pjoin, polymorphic_on=pjoin.c.type
)
eq_(
sorted([repr(x) for x in session.query(wp)]),
[
"Employee Jdoe",
"Engineer Jenn knows how to program",
"Hacker Karina 'Badass' knows how to hack",
"Manager Sally knows how to manage things",
],
)
def test_without_default_polymorphic_three(self, two_pjoin_fixture):
(
session,
Employee,
Engineer,
Manager,
Hacker,
pjoin,
pjoin2,
jdoe,
sally,
jenn,
hacker,
) = two_pjoin_fixture
eq_(
sorted([repr(x) for x in session.query(Manager)]),
["Manager Sally knows how to manage things"],
)
def test_without_default_polymorphic_four(self, two_pjoin_fixture):
(
session,
Employee,
Engineer,
Manager,
Hacker,
pjoin,
pjoin2,
jdoe,
sally,
jenn,
hacker,
) = two_pjoin_fixture
wp2 = with_polymorphic(
Engineer, "*", pjoin2, polymorphic_on=pjoin2.c.type
)
eq_(
sorted([repr(x) for x in session.query(wp2)]),
[
"Engineer Jenn knows how to program",
"Hacker Karina 'Badass' knows how to hack",
],
)
def test_without_default_polymorphic_five(self, two_pjoin_fixture):
(
session,
Employee,
Engineer,
Manager,
Hacker,
pjoin,
pjoin2,
jdoe,
sally,
jenn,
hacker,
) = two_pjoin_fixture
eq_(
[repr(x) for x in session.query(Hacker)],
["Hacker Karina 'Badass' knows how to hack"],
)
def test_without_default_polymorphic_six(self, two_pjoin_fixture):
(
session,
Employee,
Engineer,
Manager,
Hacker,
pjoin,
pjoin2,
jdoe,
sally,
jenn,
hacker,
) = two_pjoin_fixture
subq = pjoin2.select().subquery()
wp2 = with_polymorphic(Engineer, "*", subq, polymorphic_on=subq.c.type)
eq_(
sorted([repr(x) for x in session.query(wp2)]),
[
"Engineer Jenn knows how to program",
"Hacker Karina 'Badass' knows how to hack",
],
)
@testing.combinations(True, False, argnames="use_star")
def test_without_default_polymorphic_buildit_newstyle(
self, two_pjoin_fixture, use_star
):
"""how would we do these concrete polymorphic queries using 2.0 style,
and not any old and esoteric features like "polymorphic_union" ?
"""
(
session,
Employee,
Engineer,
Manager,
Hacker,
pjoin,
pjoin2,
jdoe,
sally,
jenn,
hacker,
) = two_pjoin_fixture
stmt = union_all(
select(
literal("engineer").label("type"),
Engineer,
null().label("nickname"),
),
select(literal("hacker").label("type"), Hacker),
).subquery()
if use_star:
wp = with_polymorphic(
Engineer, "*", stmt, polymorphic_on=stmt.c.type
)
else:
wp = with_polymorphic(
Engineer, [Engineer, Hacker], stmt, polymorphic_on=stmt.c.type
)
result = session.execute(select(wp)).scalars()
eq_(
sorted(repr(obj) for obj in result),
[
"Engineer Jenn knows how to program",
"Hacker Karina 'Badass' knows how to hack",
],
)
def test_relationship(self):
Employee, Engineer, Manager = self.classes(
"Employee", "Engineer", "Manager"
)
(Company,) = self.classes("Company")
(companies,) = self.tables("companies")
engineers_table, managers_table = self.tables("engineers", "managers")
pjoin = polymorphic_union(
{"manager": managers_table, "engineer": engineers_table},
"type",
"pjoin",
)
self.mapper_registry.map_imperatively(
Company,
companies,
properties={"employees": relationship(Employee)},
)
employee_mapper = self.mapper_registry.map_imperatively(
Employee, pjoin, polymorphic_on=pjoin.c.type
)
self.mapper_registry.map_imperatively(
Manager,
managers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="manager",
)
self.mapper_registry.map_imperatively(
Engineer,
engineers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="engineer",
)
session = fixture_session()
c = Company()
c.employees.append(Manager("Sally", "knows how to manage things"))
c.employees.append(Engineer("Karina", "knows how to hack"))
session.add(c)
session.flush()
session.expunge_all()
def go():
c2 = session.get(Company, c.id)
assert {repr(x) for x in c2.employees} == {
"Engineer Karina knows how to hack",
"Manager Sally knows how to manage things",
}
self.assert_sql_count(testing.db, go, 2)
session.expunge_all()
def go():
c2 = session.get(
Company, c.id, options=[joinedload(Company.employees)]
)
assert {repr(x) for x in c2.employees} == {
"Engineer Karina knows how to hack",
"Manager Sally knows how to manage things",
}
self.assert_sql_count(testing.db, go, 1)
class PropertyInheritanceTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table(
"a_table",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("some_dest_id", Integer, ForeignKey("dest_table.id")),
Column("aname", String(50)),
)
Table(
"b_table",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("some_dest_id", Integer, ForeignKey("dest_table.id")),
Column("bname", String(50)),
)
Table(
"c_table",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("some_dest_id", Integer, ForeignKey("dest_table.id")),
Column("cname", String(50)),
)
Table(
"dest_table",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(50)),
)
@classmethod
def setup_classes(cls):
class A(cls.Comparable):
pass
class B(A):
pass
class C(A):
pass
class Dest(cls.Comparable):
pass
def test_noninherited_warning(self):
A, B, b_table, a_table, Dest, dest_table = (
self.classes.A,
self.classes.B,
self.tables.b_table,
self.tables.a_table,
self.classes.Dest,
self.tables.dest_table,
)
self.mapper_registry.map_imperatively(
A, a_table, properties={"some_dest": relationship(Dest)}
)
self.mapper_registry.map_imperatively(
B, b_table, inherits=A, concrete=True
)
self.mapper_registry.map_imperatively(Dest, dest_table)
b = B()
dest = Dest()
assert_raises(AttributeError, setattr, b, "some_dest", dest)
clear_mappers()
self.mapper_registry.map_imperatively(
A, a_table, properties={"a_id": a_table.c.id}
)
self.mapper_registry.map_imperatively(
B, b_table, inherits=A, concrete=True
)
self.mapper_registry.map_imperatively(Dest, dest_table)
b = B()
assert_raises(AttributeError, setattr, b, "a_id", 3)
clear_mappers()
self.mapper_registry.map_imperatively(
A, a_table, properties={"a_id": a_table.c.id}
)
self.mapper_registry.map_imperatively(
B, b_table, inherits=A, concrete=True
)
self.mapper_registry.map_imperatively(Dest, dest_table)
def test_inheriting(self):
A, B, b_table, a_table, Dest, dest_table = (
self.classes.A,
self.classes.B,
self.tables.b_table,
self.tables.a_table,
self.classes.Dest,
self.tables.dest_table,
)
self.mapper_registry.map_imperatively(
A,
a_table,
properties={
"some_dest": relationship(Dest, back_populates="many_a")
},
)
self.mapper_registry.map_imperatively(
B,
b_table,
inherits=A,
concrete=True,
properties={
"some_dest": relationship(Dest, back_populates="many_b")
},
)
self.mapper_registry.map_imperatively(
Dest,
dest_table,
properties={
"many_a": relationship(A, back_populates="some_dest"),
"many_b": relationship(B, back_populates="some_dest"),
},
)
sess = fixture_session()
dest1 = Dest(name="c1")
dest2 = Dest(name="c2")
a1 = A(some_dest=dest1, aname="a1")
a2 = A(some_dest=dest2, aname="a2")
b1 = B(some_dest=dest1, bname="b1")
b2 = B(some_dest=dest1, bname="b2")
assert_raises(AttributeError, setattr, b1, "aname", "foo")
assert_raises(AttributeError, getattr, A, "bname")
assert dest2.many_a == [a2]
assert dest1.many_a == [a1]
assert dest1.many_b == [b1, b2]
sess.add_all([dest1, dest2])
sess.commit()
assert sess.query(Dest).filter(Dest.many_a.contains(a2)).one() is dest2
assert dest2.many_a == [a2]
assert dest1.many_a == [a1]
assert dest1.many_b == [b1, b2]
assert sess.query(B).filter(B.bname == "b1").one() is b1
def test_overlapping_backref_relationship(self):
"""test #3630.
was revisited in #4629 (not fixed until 2.0.0rc1 despite the old
issue number)
"""
A, B, b_table, a_table, Dest, dest_table = (
self.classes.A,
self.classes.B,
self.tables.b_table,
self.tables.a_table,
self.classes.Dest,
self.tables.dest_table,
)
self.mapper_registry.map_imperatively(A, a_table)
self.mapper_registry.map_imperatively(
B, b_table, inherits=A, concrete=True
)
self.mapper_registry.map_imperatively(
Dest,
dest_table,
properties={
"a": relationship(A, backref="dest"),
"a1": relationship(B, backref="dest"),
},
)
configure_mappers()
def test_overlapping_forwards_relationship(self):
A, B, b_table, a_table, Dest, dest_table = (
self.classes.A,
self.classes.B,
self.tables.b_table,
self.tables.a_table,
self.classes.Dest,
self.tables.dest_table,
)
self.mapper_registry.map_imperatively(
A, a_table, properties={"dest": relationship(Dest, backref="a")}
)
self.mapper_registry.map_imperatively(
B,
b_table,
inherits=A,
concrete=True,
properties={"dest": relationship(Dest, backref="a1")},
)
self.mapper_registry.map_imperatively(Dest, dest_table)
configure_mappers()
def test_polymorphic_backref(self):
"""test multiple backrefs to the same polymorphically-loading
attribute."""
A, C, B, c_table, b_table, a_table, Dest, dest_table = (
self.classes.A,
self.classes.C,
self.classes.B,
self.tables.c_table,
self.tables.b_table,
self.tables.a_table,
self.classes.Dest,
self.tables.dest_table,
)
ajoin = polymorphic_union(
{"a": a_table, "b": b_table, "c": c_table}, "type", "ajoin"
)
self.mapper_registry.map_imperatively(
A,
a_table,
with_polymorphic=("*", ajoin),
polymorphic_on=ajoin.c.type,
polymorphic_identity="a",
properties={
"some_dest": relationship(Dest, back_populates="many_a")
},
)
self.mapper_registry.map_imperatively(
B,
b_table,
inherits=A,
concrete=True,
polymorphic_identity="b",
properties={
"some_dest": relationship(Dest, back_populates="many_a")
},
)
self.mapper_registry.map_imperatively(
C,
c_table,
inherits=A,
concrete=True,
polymorphic_identity="c",
properties={
"some_dest": relationship(Dest, back_populates="many_a")
},
)
self.mapper_registry.map_imperatively(
Dest,
dest_table,
properties={
"many_a": relationship(
A, back_populates="some_dest", order_by=ajoin.c.id
)
},
)
sess = fixture_session()
dest1 = Dest(name="c1")
dest2 = Dest(name="c2")
a1 = A(some_dest=dest1, aname="a1", id=1)
a2 = A(some_dest=dest2, aname="a2", id=2)
b1 = B(some_dest=dest1, bname="b1", id=3)
b2 = B(some_dest=dest1, bname="b2", id=4)
c1 = C(some_dest=dest1, cname="c1", id=5)
c2 = C(some_dest=dest2, cname="c2", id=6)
eq_([a2, c2], dest2.many_a)
eq_([a1, b1, b2, c1], dest1.many_a)
sess.add_all([dest1, dest2])
sess.commit()
assert sess.query(Dest).filter(Dest.many_a.contains(a2)).one() is dest2
assert sess.query(Dest).filter(Dest.many_a.contains(b1)).one() is dest1
assert sess.query(Dest).filter(Dest.many_a.contains(c2)).one() is dest2
eq_(dest2.many_a, [a2, c2])
eq_(dest1.many_a, [a1, b1, b2, c1])
sess.expire_all()
def go():
eq_(
[
Dest(
many_a=[
A(aname="a1"),
B(bname="b1"),
B(bname="b2"),
C(cname="c1"),
]
),
Dest(many_a=[A(aname="a2"), C(cname="c2")]),
],
sess.query(Dest)
.options(joinedload(Dest.many_a))
.order_by(Dest.id)
.all(),
)
self.assert_sql_count(testing.db, go, 1)
def test_merge_w_relationship(self):
A, C, B, c_table, b_table, a_table, Dest, dest_table = (
self.classes.A,
self.classes.C,
self.classes.B,
self.tables.c_table,
self.tables.b_table,
self.tables.a_table,
self.classes.Dest,
self.tables.dest_table,
)
ajoin = polymorphic_union(
{"a": a_table, "b": b_table, "c": c_table}, "type", "ajoin"
)
self.mapper_registry.map_imperatively(
A,
a_table,
with_polymorphic=("*", ajoin),
polymorphic_on=ajoin.c.type,
polymorphic_identity="a",
properties={
"some_dest": relationship(Dest, back_populates="many_a")
},
)
self.mapper_registry.map_imperatively(
B,
b_table,
inherits=A,
concrete=True,
polymorphic_identity="b",
properties={
"some_dest": relationship(Dest, back_populates="many_a")
},
)
self.mapper_registry.map_imperatively(
C,
c_table,
inherits=A,
concrete=True,
polymorphic_identity="c",
properties={
"some_dest": relationship(Dest, back_populates="many_a")
},
)
self.mapper_registry.map_imperatively(
Dest,
dest_table,
properties={
"many_a": relationship(
A, back_populates="some_dest", order_by=ajoin.c.id
)
},
)
assert C.some_dest.property.parent is class_mapper(C)
assert B.some_dest.property.parent is class_mapper(B)
assert A.some_dest.property.parent is class_mapper(A)
sess = fixture_session()
dest1 = Dest(name="d1")
dest2 = Dest(name="d2")
a1 = A(some_dest=dest2, aname="a1")
b1 = B(some_dest=dest1, bname="b1")
c1 = C(some_dest=dest2, cname="c1")
sess.add_all([dest1, dest2, c1, a1, b1])
sess.commit()
sess2 = fixture_session()
merged_c1 = sess2.merge(c1)
eq_(merged_c1.some_dest.name, "d2")
eq_(merged_c1.some_dest_id, c1.some_dest_id)
class ManySallyanyTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table(
"base",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
)
Table(
"sub",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
)
Table(
"base_mSally",
metadata,
Column(
"base_id", Integer, ForeignKey("base.id"), primary_key=True
),
Column(
"related_id",
Integer,
ForeignKey("related.id"),
primary_key=True,
),
)
Table(
"sub_mSally",
metadata,
Column("base_id", Integer, ForeignKey("sub.id"), primary_key=True),
Column(
"related_id",
Integer,
ForeignKey("related.id"),
primary_key=True,
),
)
Table(
"related",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
)
@classmethod
def setup_classes(cls):
class Base(cls.Comparable):
pass
class Sub(Base):
pass
class Related(cls.Comparable):
pass
def test_selective_relationships(self):
sub, base_mSally, Related, Base, related, sub_mSally, base, Sub = (
self.tables.sub,
self.tables.base_mSally,
self.classes.Related,
self.classes.Base,
self.tables.related,
self.tables.sub_mSally,
self.tables.base,
self.classes.Sub,
)
self.mapper_registry.map_imperatively(
Base,
base,
properties={
"related": relationship(
Related,
secondary=base_mSally,
backref="bases",
order_by=related.c.id,
)
},
)
self.mapper_registry.map_imperatively(
Sub,
sub,
inherits=Base,
concrete=True,
properties={
"related": relationship(
Related,
secondary=sub_mSally,
backref="subs",
order_by=related.c.id,
)
},
)
self.mapper_registry.map_imperatively(Related, related)
sess = fixture_session()
b1, s1, r1, r2, r3 = Base(), Sub(), Related(), Related(), Related()
b1.related.append(r1)
b1.related.append(r2)
s1.related.append(r2)
s1.related.append(r3)
sess.add_all([b1, s1])
sess.commit()
eq_(s1.related, [r2, r3])
eq_(b1.related, [r1, r2])
class ColKeysTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
global offices_table, refugees_table
refugees_table = Table(
"refugee",
metadata,
Column(
"refugee_fid",
Integer,
primary_key=True,
test_needs_autoincrement=True,
),
Column("refugee_name", String(30), key="name"),
)
offices_table = Table(
"office",
metadata,
Column(
"office_fid",
Integer,
primary_key=True,
test_needs_autoincrement=True,
),
Column("office_name", String(30), key="name"),
)
@classmethod
def insert_data(cls, connection):
connection.execute(
refugees_table.insert(),
[
dict(refugee_fid=1, name="refugee1"),
dict(refugee_fid=2, name="refugee2"),
],
)
connection.execute(
offices_table.insert(),
[
dict(office_fid=1, name="office1"),
dict(office_fid=2, name="office2"),
],
)
def test_keys(self):
pjoin = polymorphic_union(
{"refugee": refugees_table, "office": offices_table},
"type",
"pjoin",
)
class Location:
pass
class Refugee(Location):
pass
class Office(Location):
pass
location_mapper = self.mapper_registry.map_imperatively(
Location,
pjoin,
polymorphic_on=pjoin.c.type,
polymorphic_identity="location",
)
self.mapper_registry.map_imperatively(
Office,
offices_table,
inherits=location_mapper,
concrete=True,
polymorphic_identity="office",
)
self.mapper_registry.map_imperatively(
Refugee,
refugees_table,
inherits=location_mapper,
concrete=True,
polymorphic_identity="refugee",
)
sess = fixture_session()
eq_(sess.get(Refugee, 1).name, "refugee1")
eq_(sess.get(Refugee, 2).name, "refugee2")
eq_(sess.get(Office, 1).name, "office1")
eq_(sess.get(Office, 2).name, "office2")
class AdaptOnNamesTest(
RemoveORMEventsGlobally, fixtures.DeclarativeMappedTest
):
"""test the full integration case for #7805"""
@classmethod
def setup_classes(cls):
Base = cls.DeclarativeBasic
Basic = cls.Basic
class Metadata(ComparableEntity, Base):
__tablename__ = "metadata"
id = Column(
Integer,
primary_key=True,
)
some_data = Column(String(50))
class BaseObj(ComparableEntity, AbstractConcreteBase, Base):
"""abstract concrete base with a custom polymorphic_union.
Additionally, at query time it needs to use a new version of this
union each time in order to add filter criteria. this is because
polymorphic_union() is of course very inefficient in its form
and if someone actually has to use this, it's likely better for
filter criteria to be within each sub-select. The current use
case here does not really have easy answers as we don't have
a built-in widget that does this. The complexity / little use
ratio doesn't justify it unfortunately.
This use case might be easier if we were mapped to something that
can be adapted. however, we are using adapt_on_names here as this
is usually what's more accessible to someone trying to get into
this, or at least we should make that feature work as well as it
can.
"""
strict_attrs = True
@declared_attr
def id(cls):
return Column(Integer, primary_key=True)
@declared_attr
def metadata_id(cls):
return Column(ForeignKey(Metadata.id), nullable=False)
@classmethod
def _create_polymorphic_union(cls, mappers, discriminator_name):
return cls.make_statement().subquery()
@declared_attr
def related_metadata(cls):
return relationship(Metadata)
@classmethod
def make_statement(cls, *filter_cond, include_metadata=False):
a_stmt = (
select(
A.id,
A.metadata_id,
A.thing1,
A.x1,
A.y1,
null().label("thing2"),
null().label("x2"),
null().label("y2"),
literal("a").label("type"),
)
.join(Metadata)
.filter(*filter_cond)
)
if include_metadata:
a_stmt = a_stmt.add_columns(Metadata.__table__)
b_stmt = (
select(
B.id,
B.metadata_id,
null().label("thing1"),
null().label("x1"),
null().label("y1"),
B.thing2,
B.x2,
B.y2,
literal("b").label("type"),
)
.join(Metadata)
.filter(*filter_cond)
)
if include_metadata:
b_stmt = b_stmt.add_columns(Metadata.__table__)
return union(a_stmt, b_stmt)
class XYThing(Basic):
def __init__(self, x, y):
self.x = x
self.y = y
def __composite_values__(self):
return (self.x, self.y)
def __eq__(self, other):
return (
isinstance(other, XYThing)
and other.x == self.x
and other.y == self.y
)
def __ne__(self, other):
return not self.__eq__(other)
class A(BaseObj):
__tablename__ = "a"
thing1 = Column(String(50))
comp1 = composite(
XYThing, Column("x1", Integer), Column("y1", Integer)
)
__mapper_args__ = {"polymorphic_identity": "a", "concrete": True}
class B(BaseObj):
__tablename__ = "b"
thing2 = Column(String(50))
comp2 = composite(
XYThing, Column("x2", Integer), Column("y2", Integer)
)
__mapper_args__ = {"polymorphic_identity": "b", "concrete": True}
@classmethod
def insert_data(cls, connection):
Metadata, A, B = cls.classes("Metadata", "A", "B")
XYThing = cls.classes.XYThing
with Session(connection) as sess:
sess.add_all(
[
Metadata(id=1, some_data="m1"),
Metadata(id=2, some_data="m2"),
]
)
sess.flush()
sess.add_all(
[
A(
id=5,
metadata_id=1,
thing1="thing1",
comp1=XYThing(1, 2),
),
B(
id=6,
metadata_id=2,
thing2="thing2",
comp2=XYThing(3, 4),
),
]
)
sess.commit()
def test_contains_eager(self):
Metadata, A, B = self.classes("Metadata", "A", "B")
BaseObj = self.classes.BaseObj
XYThing = self.classes.XYThing
alias = BaseObj.make_statement(
Metadata.id < 3, include_metadata=True
).subquery()
ac = with_polymorphic(
BaseObj,
[A, B],
selectable=alias,
adapt_on_names=True,
)
mt = aliased(Metadata, alias=alias)
sess = fixture_session()
with self.sql_execution_asserter() as asserter:
objects = sess.scalars(
select(ac)
.options(
contains_eager(ac.A.related_metadata.of_type(mt)),
contains_eager(ac.B.related_metadata.of_type(mt)),
)
.order_by(ac.id)
).all()
eq_(
objects,
[
A(
id=5,
metadata_id=1,
thing1="thing1",
comp1=XYThing(1, 2),
related_metadata=Metadata(id=1, some_data="m1"),
),
B(
id=6,
metadata_id=2,
thing2="thing2",
comp2=XYThing(3, 4),
related_metadata=Metadata(id=2, some_data="m2"),
),
],
)
asserter.assert_(
CompiledSQL(
"SELECT anon_1.id, anon_1.metadata_id, anon_1.type, "
"anon_1.id_1, anon_1.some_data, anon_1.thing1, "
"anon_1.x1, anon_1.y1, anon_1.thing2, anon_1.x2, anon_1.y2 "
"FROM "
"(SELECT a.id AS id, a.metadata_id AS metadata_id, "
"a.thing1 AS thing1, a.x1 AS x1, a.y1 AS y1, "
"NULL AS thing2, NULL AS x2, NULL AS y2, :param_1 AS type, "
"metadata.id AS id_1, metadata.some_data AS some_data "
"FROM a JOIN metadata ON metadata.id = a.metadata_id "
"WHERE metadata.id < :id_2 UNION SELECT b.id AS id, "
"b.metadata_id AS metadata_id, NULL AS thing1, NULL AS x1, "
"NULL AS y1, b.thing2 AS thing2, b.x2 AS x2, b.y2 AS y2, "
":param_2 AS type, metadata.id AS id_1, "
"metadata.some_data AS some_data FROM b "
"JOIN metadata ON metadata.id = b.metadata_id "
"WHERE metadata.id < :id_3) AS anon_1 ORDER BY anon_1.id",
[{"param_1": "a", "id_2": 3, "param_2": "b", "id_3": 3}],
)
)