summaryrefslogtreecommitdiff
path: root/test/orm/inheritance/test_poly_persistence.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2019-01-06 01:14:26 -0500
committermike bayer <mike_mp@zzzcomputing.com>2019-01-06 17:34:50 +0000
commit1e1a38e7801f410f244e4bbb44ec795ae152e04e (patch)
tree28e725c5c8188bd0cfd133d1e268dbca9b524978 /test/orm/inheritance/test_poly_persistence.py
parent404e69426b05a82d905cbb3ad33adafccddb00dd (diff)
downloadsqlalchemy-1e1a38e7801f410f244e4bbb44ec795ae152e04e.tar.gz
Run black -l 79 against all source files
This is a straight reformat run using black as is, with no edits applied at all. The black run will format code consistently, however in some cases that are prevalent in SQLAlchemy code it produces too-long lines. The too-long lines will be resolved in the following commit that will resolve all remaining flake8 issues including shadowed builtins, long lines, import order, unused imports, duplicate imports, and docstring issues. Change-Id: I7eda77fed3d8e73df84b3651fd6cfcfe858d4dc9
Diffstat (limited to 'test/orm/inheritance/test_poly_persistence.py')
-rw-r--r--test/orm/inheritance/test_poly_persistence.py472
1 files changed, 305 insertions, 167 deletions
diff --git a/test/orm/inheritance/test_poly_persistence.py b/test/orm/inheritance/test_poly_persistence.py
index 3713ca24f..67ae78fb4 100644
--- a/test/orm/inheritance/test_poly_persistence.py
+++ b/test/orm/inheritance/test_poly_persistence.py
@@ -37,45 +37,75 @@ class PolymorphTest(fixtures.MappedTest):
def define_tables(cls, metadata):
global companies, people, engineers, managers, boss
- companies = Table('companies', metadata,
- Column('company_id', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('name', String(50)))
+ companies = Table(
+ "companies",
+ metadata,
+ Column(
+ "company_id",
+ Integer,
+ primary_key=True,
+ test_needs_autoincrement=True,
+ ),
+ Column("name", String(50)),
+ )
people = Table(
- 'people', metadata,
+ "people",
+ metadata,
Column(
- 'person_id', Integer, primary_key=True,
- test_needs_autoincrement=True),
+ "person_id",
+ Integer,
+ primary_key=True,
+ test_needs_autoincrement=True,
+ ),
Column(
- 'company_id', Integer, ForeignKey('companies.company_id'),
- nullable=False),
- Column('name', String(50)),
- Column('type', String(30)))
+ "company_id",
+ Integer,
+ ForeignKey("companies.company_id"),
+ nullable=False,
+ ),
+ Column("name", String(50)),
+ Column("type", String(30)),
+ )
engineers = Table(
- 'engineers', metadata,
+ "engineers",
+ metadata,
Column(
- 'person_id', Integer, ForeignKey('people.person_id'),
- primary_key=True),
- Column('status', String(30)),
- Column('engineer_name', String(50)),
- Column('primary_language', String(50)))
+ "person_id",
+ Integer,
+ ForeignKey("people.person_id"),
+ primary_key=True,
+ ),
+ Column("status", String(30)),
+ Column("engineer_name", String(50)),
+ Column("primary_language", String(50)),
+ )
managers = Table(
- 'managers', metadata,
+ "managers",
+ metadata,
Column(
- 'person_id', Integer, ForeignKey('people.person_id'),
- primary_key=True),
- Column('status', String(30)),
- Column('manager_name', String(50)))
+ "person_id",
+ Integer,
+ ForeignKey("people.person_id"),
+ primary_key=True,
+ ),
+ Column("status", String(30)),
+ Column("manager_name", String(50)),
+ )
boss = Table(
- 'boss', metadata,
+ "boss",
+ metadata,
Column(
- 'boss_id', Integer, ForeignKey('managers.person_id'),
- primary_key=True),
- Column('golf_swing', String(30)))
+ "boss_id",
+ Integer,
+ ForeignKey("managers.person_id"),
+ primary_key=True,
+ ),
+ Column("golf_swing", String(30)),
+ )
metadata.create_all()
@@ -87,43 +117,73 @@ class InsertOrderTest(PolymorphTest):
person_join = polymorphic_union(
{
- 'engineer': people.join(engineers),
- 'manager': people.join(managers),
- 'person': people.select(people.c.type == 'person'),
- }, None, 'pjoin')
-
- person_mapper = mapper(Person, people,
- with_polymorphic=('*', person_join),
- polymorphic_on=person_join.c.type,
- polymorphic_identity='person')
-
- mapper(Engineer, engineers, inherits=person_mapper,
- polymorphic_identity='engineer')
- mapper(Manager, managers, inherits=person_mapper,
- polymorphic_identity='manager')
- mapper(Company, companies, properties={
- 'employees': relationship(Person,
- backref='company',
- order_by=person_join.c.person_id)
- })
+ "engineer": people.join(engineers),
+ "manager": people.join(managers),
+ "person": people.select(people.c.type == "person"),
+ },
+ None,
+ "pjoin",
+ )
+
+ person_mapper = mapper(
+ Person,
+ people,
+ with_polymorphic=("*", person_join),
+ polymorphic_on=person_join.c.type,
+ polymorphic_identity="person",
+ )
+
+ mapper(
+ Engineer,
+ engineers,
+ inherits=person_mapper,
+ polymorphic_identity="engineer",
+ )
+ mapper(
+ Manager,
+ managers,
+ inherits=person_mapper,
+ polymorphic_identity="manager",
+ )
+ mapper(
+ Company,
+ companies,
+ properties={
+ "employees": relationship(
+ Person, backref="company", order_by=person_join.c.person_id
+ )
+ },
+ )
session = create_session()
- c = Company(name='company1')
+ c = Company(name="company1")
c.employees.append(
Manager(
- status='AAB', manager_name='manager1',
- name='pointy haired boss'))
- c.employees.append(Engineer(status='BBA',
- engineer_name='engineer1',
- primary_language='java', name='dilbert'))
- c.employees.append(Person(status='HHH', name='joesmith'))
- c.employees.append(Engineer(status='CGG',
- engineer_name='engineer2',
- primary_language='python', name='wally'))
+ status="AAB",
+ manager_name="manager1",
+ name="pointy haired boss",
+ )
+ )
c.employees.append(
- Manager(
- status='ABA', manager_name='manager2',
- name='jsmith'))
+ Engineer(
+ status="BBA",
+ engineer_name="engineer1",
+ primary_language="java",
+ name="dilbert",
+ )
+ )
+ c.employees.append(Person(status="HHH", name="joesmith"))
+ c.employees.append(
+ Engineer(
+ status="CGG",
+ engineer_name="engineer2",
+ primary_language="python",
+ name="wally",
+ )
+ )
+ c.employees.append(
+ Manager(status="ABA", manager_name="manager2", name="jsmith")
+ )
session.add(c)
session.flush()
session.expunge_all()
@@ -134,8 +194,9 @@ class RoundTripTest(PolymorphTest):
pass
-def _generate_round_trip_test(include_base, lazy_relationship,
- redefine_colprop, with_polymorphic):
+def _generate_round_trip_test(
+ include_base, lazy_relationship, redefine_colprop, with_polymorphic
+):
"""generates a round trip test.
include_base - whether or not to include the base 'person' type in
@@ -151,84 +212,124 @@ def _generate_round_trip_test(include_base, lazy_relationship,
"""
def test_roundtrip(self):
- if with_polymorphic == 'unions':
+ if with_polymorphic == "unions":
if include_base:
person_join = polymorphic_union(
{
- 'engineer': people.join(engineers),
- 'manager': people.join(managers),
- 'person': people.select(people.c.type == 'person'),
- }, None, 'pjoin')
+ "engineer": people.join(engineers),
+ "manager": people.join(managers),
+ "person": people.select(people.c.type == "person"),
+ },
+ None,
+ "pjoin",
+ )
else:
person_join = polymorphic_union(
{
- 'engineer': people.join(engineers),
- 'manager': people.join(managers),
- }, None, 'pjoin')
+ "engineer": people.join(engineers),
+ "manager": people.join(managers),
+ },
+ None,
+ "pjoin",
+ )
manager_join = people.join(managers).outerjoin(boss)
- person_with_polymorphic = ['*', person_join]
- manager_with_polymorphic = ['*', manager_join]
- elif with_polymorphic == 'joins':
- person_join = people.outerjoin(engineers).outerjoin(managers).\
- outerjoin(boss)
+ person_with_polymorphic = ["*", person_join]
+ manager_with_polymorphic = ["*", manager_join]
+ elif with_polymorphic == "joins":
+ person_join = (
+ people.outerjoin(engineers).outerjoin(managers).outerjoin(boss)
+ )
manager_join = people.join(managers).outerjoin(boss)
- person_with_polymorphic = ['*', person_join]
- manager_with_polymorphic = ['*', manager_join]
- elif with_polymorphic == 'auto':
- person_with_polymorphic = '*'
- manager_with_polymorphic = '*'
+ person_with_polymorphic = ["*", person_join]
+ manager_with_polymorphic = ["*", manager_join]
+ elif with_polymorphic == "auto":
+ person_with_polymorphic = "*"
+ manager_with_polymorphic = "*"
else:
person_with_polymorphic = None
manager_with_polymorphic = None
if redefine_colprop:
- person_mapper = mapper(Person, people,
- with_polymorphic=person_with_polymorphic,
- polymorphic_on=people.c.type,
- polymorphic_identity='person',
- properties={'person_name': people.c.name})
+ person_mapper = mapper(
+ Person,
+ people,
+ with_polymorphic=person_with_polymorphic,
+ polymorphic_on=people.c.type,
+ polymorphic_identity="person",
+ properties={"person_name": people.c.name},
+ )
else:
- person_mapper = mapper(Person, people,
- with_polymorphic=person_with_polymorphic,
- polymorphic_on=people.c.type,
- polymorphic_identity='person')
-
- mapper(Engineer, engineers, inherits=person_mapper,
- polymorphic_identity='engineer')
- mapper(Manager, managers, inherits=person_mapper,
- with_polymorphic=manager_with_polymorphic,
- polymorphic_identity='manager')
-
- mapper(Boss, boss, inherits=Manager, polymorphic_identity='boss')
-
- mapper(Company, companies, properties={
- 'employees': relationship(Person, lazy=lazy_relationship,
- cascade="all, delete-orphan",
- backref="company",
- order_by=people.c.person_id)
- })
+ person_mapper = mapper(
+ Person,
+ people,
+ with_polymorphic=person_with_polymorphic,
+ polymorphic_on=people.c.type,
+ polymorphic_identity="person",
+ )
+
+ mapper(
+ Engineer,
+ engineers,
+ inherits=person_mapper,
+ polymorphic_identity="engineer",
+ )
+ mapper(
+ Manager,
+ managers,
+ inherits=person_mapper,
+ with_polymorphic=manager_with_polymorphic,
+ polymorphic_identity="manager",
+ )
+
+ mapper(Boss, boss, inherits=Manager, polymorphic_identity="boss")
+
+ mapper(
+ Company,
+ companies,
+ properties={
+ "employees": relationship(
+ Person,
+ lazy=lazy_relationship,
+ cascade="all, delete-orphan",
+ backref="company",
+ order_by=people.c.person_id,
+ )
+ },
+ )
if redefine_colprop:
- person_attribute_name = 'person_name'
+ person_attribute_name = "person_name"
else:
- person_attribute_name = 'name'
+ person_attribute_name = "name"
employees = [
- Manager(status='AAB', manager_name='manager1',
- **{person_attribute_name: 'pointy haired boss'}),
- Engineer(status='BBA', engineer_name='engineer1',
- primary_language='java',
- **{person_attribute_name: 'dilbert'}),
+ Manager(
+ status="AAB",
+ manager_name="manager1",
+ **{person_attribute_name: "pointy haired boss"}
+ ),
+ Engineer(
+ status="BBA",
+ engineer_name="engineer1",
+ primary_language="java",
+ **{person_attribute_name: "dilbert"}
+ ),
]
if include_base:
- employees.append(Person(**{person_attribute_name: 'joesmith'}))
+ employees.append(Person(**{person_attribute_name: "joesmith"}))
employees += [
- Engineer(status='CGG', engineer_name='engineer2',
- primary_language='python',
- **{person_attribute_name: 'wally'}),
- Manager(status='ABA', manager_name='manager2',
- **{person_attribute_name: 'jsmith'})
+ Engineer(
+ status="CGG",
+ engineer_name="engineer2",
+ primary_language="python",
+ **{person_attribute_name: "wally"}
+ ),
+ Manager(
+ status="ABA",
+ manager_name="manager2",
+ **{person_attribute_name: "jsmith"}
+ ),
]
pointy = employees[0]
@@ -236,7 +337,7 @@ def _generate_round_trip_test(include_base, lazy_relationship,
dilbert = employees[1]
session = create_session()
- c = Company(name='company1')
+ c = Company(name="company1")
c.employees = employees
session.add(c)
@@ -246,9 +347,12 @@ def _generate_round_trip_test(include_base, lazy_relationship,
eq_(session.query(Person).get(dilbert.person_id), dilbert)
session.expunge_all()
- eq_(session.query(Person).filter(
- Person.person_id == dilbert.person_id).one(),
- dilbert)
+ eq_(
+ session.query(Person)
+ .filter(Person.person_id == dilbert.person_id)
+ .one(),
+ dilbert,
+ )
session.expunge_all()
def go():
@@ -256,13 +360,13 @@ def _generate_round_trip_test(include_base, lazy_relationship,
eq_(cc.employees, employees)
if not lazy_relationship:
- if with_polymorphic != 'none':
+ if with_polymorphic != "none":
self.assert_sql_count(testing.db, go, 1)
else:
self.assert_sql_count(testing.db, go, 5)
else:
- if with_polymorphic != 'none':
+ if with_polymorphic != "none":
self.assert_sql_count(testing.db, go, 2)
else:
self.assert_sql_count(testing.db, go, 6)
@@ -272,21 +376,24 @@ def _generate_round_trip_test(include_base, lazy_relationship,
# in the case of the polymorphic Person query,
# the "people" selectable should be adapted to be "person_join"
eq_(
- session.query(Person).filter(
- getattr(Person, person_attribute_name) == 'dilbert'
- ).first(),
- dilbert
+ session.query(Person)
+ .filter(getattr(Person, person_attribute_name) == "dilbert")
+ .first(),
+ dilbert,
)
- assert session.query(Person).filter(
- getattr(Person, person_attribute_name) == 'dilbert'
- ).first().person_id
+ assert (
+ session.query(Person)
+ .filter(getattr(Person, person_attribute_name) == "dilbert")
+ .first()
+ .person_id
+ )
eq_(
- session.query(Engineer).filter(
- getattr(Person, person_attribute_name) == 'dilbert'
- ).first(),
- dilbert
+ session.query(Engineer)
+ .filter(getattr(Person, person_attribute_name) == "dilbert")
+ .first(),
+ dilbert,
)
# test selecting from the query, joining against
@@ -297,27 +404,36 @@ def _generate_round_trip_test(include_base, lazy_relationship,
dilbert = session.query(Person).get(dilbert.person_id)
is_(
dilbert,
- session.query(Person).filter(
- (palias.c.name == 'dilbert') &
- (palias.c.person_id == Person.person_id)).first()
+ session.query(Person)
+ .filter(
+ (palias.c.name == "dilbert")
+ & (palias.c.person_id == Person.person_id)
+ )
+ .first(),
)
is_(
dilbert,
- session.query(Engineer).filter(
- (palias.c.name == 'dilbert') &
- (palias.c.person_id == Person.person_id)).first()
+ session.query(Engineer)
+ .filter(
+ (palias.c.name == "dilbert")
+ & (palias.c.person_id == Person.person_id)
+ )
+ .first(),
)
is_(
dilbert,
- session.query(Person).filter(
- (Engineer.engineer_name == "engineer1") &
- (engineers.c.person_id == people.c.person_id)
- ).first()
+ session.query(Person)
+ .filter(
+ (Engineer.engineer_name == "engineer1")
+ & (engineers.c.person_id == people.c.person_id)
+ )
+ .first(),
)
is_(
dilbert,
- session.query(Engineer).
- filter(Engineer.engineer_name == "engineer1")[0]
+ session.query(Engineer).filter(
+ Engineer.engineer_name == "engineer1"
+ )[0],
)
session.flush()
@@ -325,61 +441,83 @@ def _generate_round_trip_test(include_base, lazy_relationship,
def go():
session.query(Person).filter(
- getattr(Person, person_attribute_name) == 'dilbert').first()
+ getattr(Person, person_attribute_name) == "dilbert"
+ ).first()
+
self.assert_sql_count(testing.db, go, 1)
session.expunge_all()
- dilbert = session.query(Person).filter(
- getattr(Person, person_attribute_name) == 'dilbert').first()
+ dilbert = (
+ session.query(Person)
+ .filter(getattr(Person, person_attribute_name) == "dilbert")
+ .first()
+ )
def go():
# assert that only primary table is queried for
# already-present-in-session
- d = session.query(Person).filter(
- getattr(Person, person_attribute_name) == 'dilbert').first()
+ d = (
+ session.query(Person)
+ .filter(getattr(Person, person_attribute_name) == "dilbert")
+ .first()
+ )
+
self.assert_sql_count(testing.db, go, 1)
# test standalone orphans
- daboss = Boss(status='BBB',
- manager_name='boss',
- golf_swing='fore',
- **{person_attribute_name: 'daboss'})
+ daboss = Boss(
+ status="BBB",
+ manager_name="boss",
+ golf_swing="fore",
+ **{person_attribute_name: "daboss"}
+ )
session.add(daboss)
assert_raises(sa_exc.DBAPIError, session.flush)
c = session.query(Company).first()
daboss.company = c
- manager_list = [e for e in c.employees
- if isinstance(e, Manager)]
+ manager_list = [e for e in c.employees if isinstance(e, Manager)]
session.flush()
session.expunge_all()
- eq_(session.query(Manager).order_by(Manager.person_id).all(),
- manager_list)
+ eq_(
+ session.query(Manager).order_by(Manager.person_id).all(),
+ manager_list,
+ )
c = session.query(Company).first()
session.delete(c)
session.flush()
- eq_(select([func.count('*')]).select_from(people).scalar(), 0)
+ eq_(select([func.count("*")]).select_from(people).scalar(), 0)
test_roundtrip = function_named(
- test_roundtrip, "test_%s%s%s_%s" % (
+ test_roundtrip,
+ "test_%s%s%s_%s"
+ % (
(lazy_relationship and "lazy" or "eager"),
(include_base and "_inclbase" or ""),
(redefine_colprop and "_redefcol" or ""),
- with_polymorphic))
+ with_polymorphic,
+ ),
+ )
setattr(RoundTripTest, test_roundtrip.__name__, test_roundtrip)
for lazy_relationship in [True, False]:
for redefine_colprop in [True, False]:
- for with_polymorphic in ['unions', 'joins', 'auto', 'none']:
- if with_polymorphic == 'unions':
+ for with_polymorphic in ["unions", "joins", "auto", "none"]:
+ if with_polymorphic == "unions":
for include_base in [True, False]:
_generate_round_trip_test(
- include_base, lazy_relationship, redefine_colprop,
- with_polymorphic)
+ include_base,
+ lazy_relationship,
+ redefine_colprop,
+ with_polymorphic,
+ )
else:
- _generate_round_trip_test(False,
- lazy_relationship,
- redefine_colprop, with_polymorphic)
+ _generate_round_trip_test(
+ False,
+ lazy_relationship,
+ redefine_colprop,
+ with_polymorphic,
+ )