diff options
| author | Mike Bayer <mike_mp@zzzcomputing.com> | 2019-01-06 01:14:26 -0500 |
|---|---|---|
| committer | mike bayer <mike_mp@zzzcomputing.com> | 2019-01-06 17:34:50 +0000 |
| commit | 1e1a38e7801f410f244e4bbb44ec795ae152e04e (patch) | |
| tree | 28e725c5c8188bd0cfd133d1e268dbca9b524978 /test/orm/inheritance/test_single.py | |
| parent | 404e69426b05a82d905cbb3ad33adafccddb00dd (diff) | |
| download | sqlalchemy-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_single.py')
| -rw-r--r-- | test/orm/inheritance/test_single.py | 1169 |
1 files changed, 711 insertions, 458 deletions
diff --git a/test/orm/inheritance/test_single.py b/test/orm/inheritance/test_single.py index 2416fdc29..13e1937a4 100644 --- a/test/orm/inheritance/test_single.py +++ b/test/orm/inheritance/test_single.py @@ -8,24 +8,39 @@ from sqlalchemy.testing import fixtures, AssertsCompiledSQL from sqlalchemy.testing.schema import Table, Column from sqlalchemy import inspect + class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): - __dialect__ = 'default' + __dialect__ = "default" @classmethod def define_tables(cls, metadata): - Table('employees', metadata, - Column('employee_id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('name', String(50)), - Column('manager_data', String(50)), - Column('engineer_info', String(50)), - Column('type', String(20))) - - Table('reports', metadata, - Column('report_id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('employee_id', ForeignKey('employees.employee_id')), - Column('name', String(50)),) + Table( + "employees", + metadata, + Column( + "employee_id", + Integer, + primary_key=True, + test_needs_autoincrement=True, + ), + Column("name", String(50)), + Column("manager_data", String(50)), + Column("engineer_info", String(50)), + Column("type", String(20)), + ) + + Table( + "reports", + metadata, + Column( + "report_id", + Integer, + primary_key=True, + test_needs_autoincrement=True, + ), + Column("employee_id", ForeignKey("employees.employee_id")), + Column("name", String(50)), + ) @classmethod def setup_classes(cls): @@ -46,27 +61,35 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): @classmethod def setup_mappers(cls): Employee, Manager, JuniorEngineer, employees, Engineer = ( - cls.classes.Employee, cls.classes.Manager, cls.classes. - JuniorEngineer, cls.tables.employees, cls.classes.Engineer) + cls.classes.Employee, + cls.classes.Manager, + cls.classes.JuniorEngineer, + cls.tables.employees, + cls.classes.Engineer, + ) mapper(Employee, employees, polymorphic_on=employees.c.type) - mapper(Manager, inherits=Employee, polymorphic_identity='manager') - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') - mapper(JuniorEngineer, inherits=Engineer, - polymorphic_identity='juniorengineer') + mapper(Manager, inherits=Employee, polymorphic_identity="manager") + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") + mapper( + JuniorEngineer, + inherits=Engineer, + polymorphic_identity="juniorengineer", + ) def _fixture_one(self): Employee, JuniorEngineer, Manager, Engineer = ( self.classes.Employee, self.classes.JuniorEngineer, self.classes.Manager, - self.classes.Engineer) + self.classes.Engineer, + ) session = create_session() - m1 = Manager(name='Tom', manager_data='knows how to manage things') - e1 = Engineer(name='Kurt', engineer_info='knows how to hack') - e2 = JuniorEngineer(name='Ed', engineer_info='oh that ed') + m1 = Manager(name="Tom", manager_data="knows how to manage things") + e1 = Engineer(name="Kurt", engineer_info="knows how to hack") + e2 = JuniorEngineer(name="Ed", engineer_info="oh that ed") session.add_all([m1, e1, e2]) session.flush() return session, m1, e1, e2 @@ -76,7 +99,8 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): self.classes.Employee, self.classes.JuniorEngineer, self.classes.Manager, - self.classes.Engineer) + self.classes.Engineer, + ) session, m1, e1, e2 = self._fixture_one() @@ -86,50 +110,51 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): assert session.query(JuniorEngineer).all() == [e2] m1 = session.query(Manager).one() - session.expire(m1, ['manager_data']) + session.expire(m1, ["manager_data"]) eq_(m1.manager_data, "knows how to manage things") - row = session.query(Engineer.name, Engineer.employee_id).filter( - Engineer.name == 'Kurt').first() - assert row.name == 'Kurt' + row = ( + session.query(Engineer.name, Engineer.employee_id) + .filter(Engineer.name == "Kurt") + .first() + ) + assert row.name == "Kurt" assert row.employee_id == e1.employee_id def test_multi_qualification(self): JuniorEngineer, Manager, Engineer = ( self.classes.JuniorEngineer, self.classes.Manager, - self.classes.Engineer) + self.classes.Engineer, + ) session, m1, e1, e2 = self._fixture_one() ealias = aliased(Engineer) - eq_( - session.query(Manager, ealias).all(), - [(m1, e1), (m1, e2)] - ) + eq_(session.query(Manager, ealias).all(), [(m1, e1), (m1, e2)]) - eq_( - session.query(Manager.name).all(), - [("Tom",)] - ) + eq_(session.query(Manager.name).all(), [("Tom",)]) eq_( session.query(Manager.name, ealias.name).all(), - [("Tom", "Kurt"), ("Tom", "Ed")] + [("Tom", "Kurt"), ("Tom", "Ed")], ) - eq_(session.query(func.upper(Manager.name), - func.upper(ealias.name)).all(), - [("TOM", "KURT"), ("TOM", "ED")]) + eq_( + session.query( + func.upper(Manager.name), func.upper(ealias.name) + ).all(), + [("TOM", "KURT"), ("TOM", "ED")], + ) eq_( session.query(Manager).add_entity(ealias).all(), - [(m1, e1), (m1, e2)] + [(m1, e1), (m1, e2)], ) eq_( session.query(Manager.name).add_column(ealias.name).all(), - [("Tom", "Kurt"), ("Tom", "Ed")] + [("Tom", "Kurt"), ("Tom", "Ed")], ) # TODO: I think raise error on this for now @@ -144,7 +169,8 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): self.classes.Employee, self.classes.JuniorEngineer, self.classes.Manager, - self.classes.Engineer) + self.classes.Engineer, + ) session, m1, e1, e2 = self._fixture_one() @@ -153,38 +179,25 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): def scalar(q): return [x for x, in q] - eq_( - scalar(session.query(Employee.employee_id)), - [m1id, e1id, e2id] - ) + eq_(scalar(session.query(Employee.employee_id)), [m1id, e1id, e2id]) - eq_( - scalar(session.query(Engineer.employee_id)), - [e1id, e2id] - ) + eq_(scalar(session.query(Engineer.employee_id)), [e1id, e2id]) - eq_( - scalar(session.query(Manager.employee_id)), [m1id] - ) + eq_(scalar(session.query(Manager.employee_id)), [m1id]) # this currently emits "WHERE type IN (?, ?) AND type IN (?, ?)", # so no result. - eq_( - session.query(Manager.employee_id, Engineer.employee_id).all(), - [] - ) + eq_(session.query(Manager.employee_id, Engineer.employee_id).all(), []) - eq_( - scalar(session.query(JuniorEngineer.employee_id)), - [e2id] - ) + eq_(scalar(session.query(JuniorEngineer.employee_id)), [e2id]) def test_bundle_qualification(self): Employee, JuniorEngineer, Manager, Engineer = ( self.classes.Employee, self.classes.JuniorEngineer, self.classes.Manager, - self.classes.Engineer) + self.classes.Engineer, + ) session, m1, e1, e2 = self._fixture_one() @@ -195,17 +208,15 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): eq_( scalar(session.query(Bundle("name", Employee.employee_id))), - [m1id, e1id, e2id] + [m1id, e1id, e2id], ) eq_( scalar(session.query(Bundle("name", Engineer.employee_id))), - [e1id, e2id] + [e1id, e2id], ) - eq_( - scalar(session.query(Bundle("name", Manager.employee_id))), [m1id] - ) + eq_(scalar(session.query(Bundle("name", Manager.employee_id))), [m1id]) # this currently emits "WHERE type IN (?, ?) AND type IN (?, ?)", # so no result. @@ -213,39 +224,41 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): session.query( Bundle("name", Manager.employee_id, Engineer.employee_id) ).all(), - [] + [], ) eq_( scalar(session.query(Bundle("name", JuniorEngineer.employee_id))), - [e2id] + [e2id], ) def test_from_self(self): Engineer = self.classes.Engineer sess = create_session() - self.assert_compile(sess.query(Engineer).from_self(), - 'SELECT anon_1.employees_employee_id AS ' - 'anon_1_employees_employee_id, ' - 'anon_1.employees_name AS ' - 'anon_1_employees_name, ' - 'anon_1.employees_manager_data AS ' - 'anon_1_employees_manager_data, ' - 'anon_1.employees_engineer_info AS ' - 'anon_1_employees_engineer_info, ' - 'anon_1.employees_type AS ' - 'anon_1_employees_type FROM (SELECT ' - 'employees.employee_id AS ' - 'employees_employee_id, employees.name AS ' - 'employees_name, employees.manager_data AS ' - 'employees_manager_data, ' - 'employees.engineer_info AS ' - 'employees_engineer_info, employees.type ' - 'AS employees_type FROM employees WHERE ' - 'employees.type IN (:type_1, :type_2)) AS ' - 'anon_1', - use_default_dialect=True) + self.assert_compile( + sess.query(Engineer).from_self(), + "SELECT anon_1.employees_employee_id AS " + "anon_1_employees_employee_id, " + "anon_1.employees_name AS " + "anon_1_employees_name, " + "anon_1.employees_manager_data AS " + "anon_1_employees_manager_data, " + "anon_1.employees_engineer_info AS " + "anon_1_employees_engineer_info, " + "anon_1.employees_type AS " + "anon_1_employees_type FROM (SELECT " + "employees.employee_id AS " + "employees_employee_id, employees.name AS " + "employees_name, employees.manager_data AS " + "employees_manager_data, " + "employees.engineer_info AS " + "employees_engineer_info, employees.type " + "AS employees_type FROM employees WHERE " + "employees.type IN (:type_1, :type_2)) AS " + "anon_1", + use_default_dialect=True, + ) def test_select_from_aliased_w_subclass(self): Engineer = self.classes.Engineer @@ -261,17 +274,17 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): ) self.assert_compile( - sess.query(literal('1')).select_from(a1), + sess.query(literal("1")).select_from(a1), "SELECT :param_1 AS param_1 FROM employees AS employees_1 " - "WHERE employees_1.type IN (:type_1, :type_2)" + "WHERE employees_1.type IN (:type_1, :type_2)", ) def test_union_modifiers(self): Engineer, Manager = self.classes("Engineer", "Manager") sess = create_session() - q1 = sess.query(Engineer).filter(Engineer.engineer_info == 'foo') - q2 = sess.query(Manager).filter(Manager.manager_data == 'bar') + q1 = sess.query(Engineer).filter(Engineer.engineer_info == "foo") + q2 = sess.query(Manager).filter(Manager.manager_data == "bar") assert_sql = ( "SELECT anon_1.employees_employee_id AS " @@ -294,7 +307,8 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): "employees.engineer_info AS employees_engineer_info, " "employees.type AS employees_type FROM employees " "WHERE employees.manager_data = :manager_data_1 " - "AND employees.type IN (:type_3)) AS anon_1") + "AND employees.type IN (:type_3)) AS anon_1" + ) for meth, token in [ (q1.union, "UNION"), @@ -308,58 +322,61 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): meth(q2), assert_sql % {"token": token}, checkparams={ - 'manager_data_1': 'bar', - 'type_2': 'juniorengineer', - 'type_3': 'manager', - 'engineer_info_1': 'foo', - 'type_1': 'engineer'}, + "manager_data_1": "bar", + "type_2": "juniorengineer", + "type_3": "manager", + "engineer_info_1": "foo", + "type_1": "engineer", + }, ) def test_from_self_count(self): Engineer = self.classes.Engineer sess = create_session() - col = func.count(literal_column('*')) + col = func.count(literal_column("*")) self.assert_compile( sess.query(Engineer.employee_id).from_self(col), "SELECT count(*) AS count_1 " "FROM (SELECT employees.employee_id AS employees_employee_id " "FROM employees " "WHERE employees.type IN (:type_1, :type_2)) AS anon_1", - use_default_dialect=True + use_default_dialect=True, ) def test_select_from_count(self): Manager, Engineer = (self.classes.Manager, self.classes.Engineer) sess = create_session() - m1 = Manager(name='Tom', manager_data='data1') - e1 = Engineer(name='Kurt', engineer_info='knows how to hack') + m1 = Manager(name="Tom", manager_data="data1") + e1 = Engineer(name="Kurt", engineer_info="knows how to hack") sess.add_all([m1, e1]) sess.flush() - eq_( - sess.query(func.count(1)).select_from(Manager).all(), - [(1, )] - ) + eq_(sess.query(func.count(1)).select_from(Manager).all(), [(1,)]) def test_select_from_subquery(self): Manager, JuniorEngineer, employees, Engineer = ( self.classes.Manager, self.classes.JuniorEngineer, self.tables.employees, - self.classes.Engineer) + self.classes.Engineer, + ) sess = create_session() - m1 = Manager(name='Tom', manager_data='data1') - m2 = Manager(name='Tom2', manager_data='data2') - e1 = Engineer(name='Kurt', engineer_info='knows how to hack') - e2 = JuniorEngineer(name='Ed', engineer_info='oh that ed') + m1 = Manager(name="Tom", manager_data="data1") + m2 = Manager(name="Tom2", manager_data="data2") + e1 = Engineer(name="Kurt", engineer_info="knows how to hack") + e2 = JuniorEngineer(name="Ed", engineer_info="oh that ed") sess.add_all([m1, m2, e1, e2]) sess.flush() - eq_(sess.query(Manager).select_from( - employees.select().limit(10)).all(), [m1, m2]) + eq_( + sess.query(Manager) + .select_from(employees.select().limit(10)) + .all(), + [m1, m2], + ) def test_count(self): Employee = self.classes.Employee @@ -368,10 +385,10 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): Engineer = self.classes.Engineer sess = create_session() - m1 = Manager(name='Tom', manager_data='data1') - m2 = Manager(name='Tom2', manager_data='data2') - e1 = Engineer(name='Kurt', engineer_info='data3') - e2 = JuniorEngineer(name='marvin', engineer_info='data4') + m1 = Manager(name="Tom", manager_data="data1") + m2 = Manager(name="Tom2", manager_data="data2") + e1 = Engineer(name="Kurt", engineer_info="data3") + e2 = JuniorEngineer(name="marvin", engineer_info="data4") sess.add_all([m1, m2, e1, e2]) sess.flush() @@ -379,8 +396,8 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): eq_(sess.query(Engineer).count(), 2) eq_(sess.query(Employee).count(), 4) - eq_(sess.query(Manager).filter(Manager.name.like('%m%')).count(), 2) - eq_(sess.query(Employee).filter(Employee.name.like('%m%')).count(), 3) + eq_(sess.query(Manager).filter(Manager.name.like("%m%")).count(), 2) + eq_(sess.query(Employee).filter(Employee.name.like("%m%")).count(), 3) def test_exists_standalone(self): Engineer = self.classes.Engineer @@ -389,50 +406,63 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): self.assert_compile( sess.query( - sess.query(Engineer).filter(Engineer.name == 'foo').exists()), + sess.query(Engineer).filter(Engineer.name == "foo").exists() + ), "SELECT EXISTS (SELECT 1 FROM employees WHERE " "employees.name = :name_1 AND employees.type " - "IN (:type_1, :type_2)) AS anon_1" + "IN (:type_1, :type_2)) AS anon_1", ) def test_type_filtering(self): - Employee, Manager, reports, Engineer = (self.classes.Employee, - self.classes.Manager, - self.tables.reports, - self.classes.Engineer) + Employee, Manager, reports, Engineer = ( + self.classes.Employee, + self.classes.Manager, + self.tables.reports, + self.classes.Engineer, + ) class Report(fixtures.ComparableEntity): pass - mapper(Report, reports, properties={ - 'employee': relationship(Employee, backref='reports')}) + mapper( + Report, + reports, + properties={"employee": relationship(Employee, backref="reports")}, + ) sess = create_session() - m1 = Manager(name='Tom', manager_data='data1') + m1 = Manager(name="Tom", manager_data="data1") r1 = Report(employee=m1) sess.add_all([m1, r1]) sess.flush() rq = sess.query(Report) - assert len(rq.filter(Report.employee.of_type(Manager).has()) - .all()) == 1 - assert len(rq.filter(Report.employee.of_type(Engineer).has()) - .all()) == 0 + assert ( + len(rq.filter(Report.employee.of_type(Manager).has()).all()) == 1 + ) + assert ( + len(rq.filter(Report.employee.of_type(Engineer).has()).all()) == 0 + ) def test_type_joins(self): - Employee, Manager, reports, Engineer = (self.classes.Employee, - self.classes.Manager, - self.tables.reports, - self.classes.Engineer) + Employee, Manager, reports, Engineer = ( + self.classes.Employee, + self.classes.Manager, + self.tables.reports, + self.classes.Engineer, + ) class Report(fixtures.ComparableEntity): pass - mapper(Report, reports, properties={ - 'employee': relationship(Employee, backref='reports')}) + mapper( + Report, + reports, + properties={"employee": relationship(Employee, backref="reports")}, + ) sess = create_session() - m1 = Manager(name='Tom', manager_data='data1') + m1 = Manager(name="Tom", manager_data="data1") r1 = Report(employee=m1) sess.add_all([m1, r1]) sess.flush() @@ -444,20 +474,29 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest): class RelationshipFromSingleTest( - testing.AssertsCompiledSQL, fixtures.MappedTest): + testing.AssertsCompiledSQL, fixtures.MappedTest +): @classmethod def define_tables(cls, metadata): - Table('employee', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('name', String(50)), - Column('type', String(20))) - - Table('employee_stuff', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('employee_id', Integer, ForeignKey('employee.id')), - Column('name', String(50))) + Table( + "employee", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("name", String(50)), + Column("type", String(20)), + ) + + Table( + "employee_stuff", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("employee_id", Integer, ForeignKey("employee.id")), + Column("name", String(50)), + ) @classmethod def setup_classes(cls): @@ -472,56 +511,91 @@ class RelationshipFromSingleTest( def test_subquery_load(self): employee, employee_stuff, Employee, Stuff, Manager = ( - self.tables.employee, self.tables.employee_stuff, self.classes. - Employee, self.classes.Stuff, self.classes.Manager) + self.tables.employee, + self.tables.employee_stuff, + self.classes.Employee, + self.classes.Stuff, + self.classes.Manager, + ) - mapper(Employee, employee, polymorphic_on=employee.c.type, - polymorphic_identity='employee') - mapper(Manager, inherits=Employee, polymorphic_identity='manager', - properties={'stuff': relationship(Stuff)}) + mapper( + Employee, + employee, + polymorphic_on=employee.c.type, + polymorphic_identity="employee", + ) + mapper( + Manager, + inherits=Employee, + polymorphic_identity="manager", + properties={"stuff": relationship(Stuff)}, + ) mapper(Stuff, employee_stuff) sess = create_session() - context = sess.query(Manager).options( - subqueryload('stuff'))._compile_context() - subq = context.attributes[('subquery', (class_mapper( - Manager), class_mapper(Manager).attrs.stuff))] - - self.assert_compile(subq, - 'SELECT employee_stuff.id AS ' - 'employee_stuff_id, employee_stuff.employee' - '_id AS employee_stuff_employee_id, ' - 'employee_stuff.name AS ' - 'employee_stuff_name, anon_1.employee_id ' - 'AS anon_1_employee_id FROM (SELECT ' - 'employee.id AS employee_id FROM employee ' - 'WHERE employee.type IN (:type_1)) AS anon_1 ' - 'JOIN employee_stuff ON anon_1.employee_id ' - '= employee_stuff.employee_id ORDER BY ' - 'anon_1.employee_id', - use_default_dialect=True) + context = ( + sess.query(Manager) + .options(subqueryload("stuff")) + ._compile_context() + ) + subq = context.attributes[ + ( + "subquery", + (class_mapper(Manager), class_mapper(Manager).attrs.stuff), + ) + ] + + self.assert_compile( + subq, + "SELECT employee_stuff.id AS " + "employee_stuff_id, employee_stuff.employee" + "_id AS employee_stuff_employee_id, " + "employee_stuff.name AS " + "employee_stuff_name, anon_1.employee_id " + "AS anon_1_employee_id FROM (SELECT " + "employee.id AS employee_id FROM employee " + "WHERE employee.type IN (:type_1)) AS anon_1 " + "JOIN employee_stuff ON anon_1.employee_id " + "= employee_stuff.employee_id ORDER BY " + "anon_1.employee_id", + use_default_dialect=True, + ) class RelationshipToSingleTest( - testing.AssertsCompiledSQL, fixtures.MappedTest): - __dialect__ = 'default' + testing.AssertsCompiledSQL, fixtures.MappedTest +): + __dialect__ = "default" @classmethod def define_tables(cls, metadata): - Table('employees', metadata, - Column('employee_id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('name', String(50)), - Column('manager_data', String(50)), - Column('engineer_info', String(50)), - Column('type', String(20)), - Column('company_id', Integer, - ForeignKey('companies.company_id'))) - - Table('companies', metadata, - Column('company_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("manager_data", String(50)), + Column("engineer_info", String(50)), + Column("type", String(20)), + Column("company_id", Integer, ForeignKey("companies.company_id")), + ) + + Table( + "companies", + metadata, + Column( + "company_id", + Integer, + primary_key=True, + test_needs_autoincrement=True, + ), + Column("name", String(50)), + ) @classmethod def setup_classes(cls): @@ -541,64 +615,85 @@ class RelationshipToSingleTest( pass def test_of_type(self): - JuniorEngineer, Company, companies, Manager,\ - Employee, employees, Engineer = (self.classes.JuniorEngineer, - self.classes.Company, - self.tables.companies, - self.classes.Manager, - self.classes.Employee, - self.tables.employees, - self.classes.Engineer) - - mapper(Company, companies, properties={ - 'employees': relationship(Employee, backref='company') - }) + JuniorEngineer, Company, companies, Manager, Employee, employees, Engineer = ( + self.classes.JuniorEngineer, + self.classes.Company, + self.tables.companies, + self.classes.Manager, + self.classes.Employee, + self.tables.employees, + self.classes.Engineer, + ) + + mapper( + Company, + companies, + properties={ + "employees": relationship(Employee, backref="company") + }, + ) mapper(Employee, employees, polymorphic_on=employees.c.type) - mapper(Manager, inherits=Employee, polymorphic_identity='manager') - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') - mapper(JuniorEngineer, inherits=Engineer, - polymorphic_identity='juniorengineer') + mapper(Manager, inherits=Employee, polymorphic_identity="manager") + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") + mapper( + JuniorEngineer, + inherits=Engineer, + polymorphic_identity="juniorengineer", + ) sess = sessionmaker()() - c1 = Company(name='c1') - c2 = Company(name='c2') + c1 = Company(name="c1") + c2 = Company(name="c2") - m1 = Manager(name='Tom', manager_data='data1', company=c1) - m2 = Manager(name='Tom2', manager_data='data2', company=c2) - e1 = Engineer(name='Kurt', engineer_info='knows how to hack', - company=c2) - e2 = JuniorEngineer(name='Ed', engineer_info='oh that ed', company=c1) + m1 = Manager(name="Tom", manager_data="data1", company=c1) + m2 = Manager(name="Tom2", manager_data="data2", company=c2) + e1 = Engineer( + name="Kurt", engineer_info="knows how to hack", company=c2 + ) + e2 = JuniorEngineer(name="Ed", engineer_info="oh that ed", company=c1) sess.add_all([c1, c2, m1, m2, e1, e2]) sess.commit() sess.expunge_all() - eq_(sess.query(Company).filter(Company.employees.of_type( - JuniorEngineer).any()).all(), [Company(name='c1'), ]) + eq_( + sess.query(Company) + .filter(Company.employees.of_type(JuniorEngineer).any()) + .all(), + [Company(name="c1")], + ) - eq_(sess.query(Company).join(Company.employees.of_type( - JuniorEngineer)).all(), [Company(name='c1'), ]) + eq_( + sess.query(Company) + .join(Company.employees.of_type(JuniorEngineer)) + .all(), + [Company(name="c1")], + ) def test_of_type_aliased_fromjoinpoint(self): - Company, Employee, Engineer = (self.classes.Company, - self.classes.Employee, - self.classes.Engineer) + Company, Employee, Engineer = ( + self.classes.Company, + self.classes.Employee, + self.classes.Engineer, + ) companies, employees = self.tables.companies, self.tables.employees - mapper(Company, companies, properties={ - 'employee': relationship(Employee) - }) + mapper( + Company, companies, properties={"employee": relationship(Employee)} + ) mapper(Employee, employees, polymorphic_on=employees.c.type) - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") sess = create_session() self.assert_compile( sess.query(Company).outerjoin( Company.employee.of_type(Engineer), - aliased=True, from_joinpoint=True), + aliased=True, + from_joinpoint=True, + ), "SELECT companies.company_id AS companies_company_id, " "companies.name AS companies_name FROM companies " "LEFT OUTER JOIN employees AS employees_1 ON " "companies.company_id = employees_1.company_id " - "AND employees_1.type IN (:type_1)" + "AND employees_1.type IN (:type_1)", ) def test_join_explicit_onclause_no_discriminator(self): @@ -606,37 +701,45 @@ class RelationshipToSingleTest( Company, Employee, Engineer = ( self.classes.Company, self.classes.Employee, - self.classes.Engineer) + self.classes.Engineer, + ) companies, employees = self.tables.companies, self.tables.employees - mapper(Company, companies, properties={ - 'employees': relationship(Employee) - }) + mapper( + Company, + companies, + properties={"employees": relationship(Employee)}, + ) mapper(Employee, employees) mapper(Engineer, inherits=Employee) sess = create_session() self.assert_compile( sess.query(Company, Engineer.name).join( - Engineer, Company.company_id == Engineer.company_id), + Engineer, Company.company_id == Engineer.company_id + ), "SELECT companies.company_id AS companies_company_id, " "companies.name AS companies_name, " "employees.name AS employees_name " "FROM companies JOIN " - "employees ON companies.company_id = employees.company_id" + "employees ON companies.company_id = employees.company_id", ) def test_outer_join_prop(self): - Company, Employee, Engineer = (self.classes.Company, - self.classes.Employee, - self.classes.Engineer) + Company, Employee, Engineer = ( + self.classes.Company, + self.classes.Employee, + self.classes.Engineer, + ) companies, employees = self.tables.companies, self.tables.employees - mapper(Company, companies, properties={ - 'engineers': relationship(Engineer) - }) + mapper( + Company, + companies, + properties={"engineers": relationship(Engineer)}, + ) mapper(Employee, employees, polymorphic_on=employees.c.type) - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") sess = create_session() self.assert_compile( @@ -645,47 +748,59 @@ class RelationshipToSingleTest( "companies.name AS companies_name, " "employees.name AS employees_name " "FROM companies LEFT OUTER JOIN employees ON companies.company_id " - "= employees.company_id AND employees.type IN (:type_1)") + "= employees.company_id AND employees.type IN (:type_1)", + ) def test_outer_join_prop_alias(self): - Company, Employee, Engineer = (self.classes.Company, - self.classes.Employee, - self.classes.Engineer) + Company, Employee, Engineer = ( + self.classes.Company, + self.classes.Employee, + self.classes.Engineer, + ) companies, employees = self.tables.companies, self.tables.employees - mapper(Company, companies, properties={ - 'engineers': relationship(Engineer) - }) + mapper( + Company, + companies, + properties={"engineers": relationship(Engineer)}, + ) mapper(Employee, employees, polymorphic_on=employees.c.type) - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") eng_alias = aliased(Engineer) sess = create_session() self.assert_compile( sess.query(Company, eng_alias.name).outerjoin( - eng_alias, Company.engineers), + eng_alias, Company.engineers + ), "SELECT companies.company_id AS companies_company_id, " "companies.name AS companies_name, employees_1.name AS " "employees_1_name FROM companies LEFT OUTER " "JOIN employees AS employees_1 ON companies.company_id " - "= employees_1.company_id AND employees_1.type IN (:type_1)") + "= employees_1.company_id AND employees_1.type IN (:type_1)", + ) def test_outer_join_literal_onclause(self): - Company, Employee, Engineer = (self.classes.Company, - self.classes.Employee, - self.classes.Engineer) + Company, Employee, Engineer = ( + self.classes.Company, + self.classes.Employee, + self.classes.Engineer, + ) companies, employees = self.tables.companies, self.tables.employees - mapper(Company, companies, properties={ - 'engineers': relationship(Engineer) - }) + mapper( + Company, + companies, + properties={"engineers": relationship(Engineer)}, + ) mapper(Employee, employees, polymorphic_on=employees.c.type) - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") sess = create_session() self.assert_compile( sess.query(Company, Engineer).outerjoin( - Engineer, Company.company_id == Engineer.company_id), + Engineer, Company.company_id == Engineer.company_id + ), "SELECT companies.company_id AS companies_company_id, " "companies.name AS companies_name, " "employees.employee_id AS employees_employee_id, " @@ -696,26 +811,31 @@ class RelationshipToSingleTest( "employees.company_id AS employees_company_id FROM companies " "LEFT OUTER JOIN employees ON " "companies.company_id = employees.company_id " - "AND employees.type IN (:type_1)" + "AND employees.type IN (:type_1)", ) def test_outer_join_literal_onclause_alias(self): - Company, Employee, Engineer = (self.classes.Company, - self.classes.Employee, - self.classes.Engineer) + Company, Employee, Engineer = ( + self.classes.Company, + self.classes.Employee, + self.classes.Engineer, + ) companies, employees = self.tables.companies, self.tables.employees - mapper(Company, companies, properties={ - 'engineers': relationship(Engineer) - }) + mapper( + Company, + companies, + properties={"engineers": relationship(Engineer)}, + ) mapper(Employee, employees, polymorphic_on=employees.c.type) - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") eng_alias = aliased(Engineer) sess = create_session() self.assert_compile( sess.query(Company, eng_alias).outerjoin( - eng_alias, Company.company_id == eng_alias.company_id), + eng_alias, Company.company_id == eng_alias.company_id + ), "SELECT companies.company_id AS companies_company_id, " "companies.name AS companies_name, " "employees_1.employee_id AS employees_1_employee_id, " @@ -726,25 +846,28 @@ class RelationshipToSingleTest( "employees_1.company_id AS employees_1_company_id " "FROM companies LEFT OUTER JOIN employees AS employees_1 ON " "companies.company_id = employees_1.company_id " - "AND employees_1.type IN (:type_1)" + "AND employees_1.type IN (:type_1)", ) def test_outer_join_no_onclause(self): - Company, Employee, Engineer = (self.classes.Company, - self.classes.Employee, - self.classes.Engineer) + Company, Employee, Engineer = ( + self.classes.Company, + self.classes.Employee, + self.classes.Engineer, + ) companies, employees = self.tables.companies, self.tables.employees - mapper(Company, companies, properties={ - 'engineers': relationship(Engineer) - }) + mapper( + Company, + companies, + properties={"engineers": relationship(Engineer)}, + ) mapper(Employee, employees, polymorphic_on=employees.c.type) - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") sess = create_session() self.assert_compile( - sess.query(Company, Engineer).outerjoin( - Engineer), + sess.query(Company, Engineer).outerjoin(Engineer), "SELECT companies.company_id AS companies_company_id, " "companies.name AS companies_name, " "employees.employee_id AS employees_employee_id, " @@ -755,26 +878,29 @@ class RelationshipToSingleTest( "employees.company_id AS employees_company_id " "FROM companies LEFT OUTER JOIN employees ON " "companies.company_id = employees.company_id " - "AND employees.type IN (:type_1)" + "AND employees.type IN (:type_1)", ) def test_outer_join_no_onclause_alias(self): - Company, Employee, Engineer = (self.classes.Company, - self.classes.Employee, - self.classes.Engineer) + Company, Employee, Engineer = ( + self.classes.Company, + self.classes.Employee, + self.classes.Engineer, + ) companies, employees = self.tables.companies, self.tables.employees - mapper(Company, companies, properties={ - 'engineers': relationship(Engineer) - }) + mapper( + Company, + companies, + properties={"engineers": relationship(Engineer)}, + ) mapper(Employee, employees, polymorphic_on=employees.c.type) - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") eng_alias = aliased(Engineer) sess = create_session() self.assert_compile( - sess.query(Company, eng_alias).outerjoin( - eng_alias), + sess.query(Company, eng_alias).outerjoin(eng_alias), "SELECT companies.company_id AS companies_company_id, " "companies.name AS companies_name, " "employees_1.employee_id AS employees_1_employee_id, " @@ -785,31 +911,34 @@ class RelationshipToSingleTest( "employees_1.company_id AS employees_1_company_id " "FROM companies LEFT OUTER JOIN employees AS employees_1 ON " "companies.company_id = employees_1.company_id " - "AND employees_1.type IN (:type_1)" + "AND employees_1.type IN (:type_1)", ) def test_correlated_column_select(self): - Company, Employee, Engineer = (self.classes.Company, - self.classes.Employee, - self.classes.Engineer) + Company, Employee, Engineer = ( + self.classes.Company, + self.classes.Employee, + self.classes.Engineer, + ) companies, employees = self.tables.companies, self.tables.employees mapper(Company, companies) mapper( - Employee, employees, + Employee, + employees, polymorphic_on=employees.c.type, - properties={ - 'company': relationship(Company) - } + properties={"company": relationship(Company)}, ) - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") sess = create_session() - engineer_count = sess.query(func.count(Engineer.employee_id)) \ - .select_from(Engineer) \ - .filter(Engineer.company_id == Company.company_id) \ - .correlate(Company) \ + engineer_count = ( + sess.query(func.count(Engineer.employee_id)) + .select_from(Engineer) + .filter(Engineer.company_id == Company.company_id) + .correlate(Company) .as_scalar() + ) self.assert_compile( sess.query(Company.company_id, engineer_count), @@ -817,47 +946,63 @@ class RelationshipToSingleTest( "(SELECT count(employees.employee_id) AS count_1 " "FROM employees WHERE employees.company_id = " "companies.company_id AND employees.type IN (:type_1)) AS anon_1 " - "FROM companies" + "FROM companies", ) def test_no_aliasing_from_overlap(self): # test [ticket:3233] - Company, Employee, Engineer, Manager = (self.classes.Company, - self.classes.Employee, - self.classes.Engineer, - self.classes.Manager) + Company, Employee, Engineer, Manager = ( + self.classes.Company, + self.classes.Employee, + self.classes.Engineer, + self.classes.Manager, + ) companies, employees = self.tables.companies, self.tables.employees - mapper(Company, companies, properties={ - 'employees': relationship(Employee, backref="company") - }) + mapper( + Company, + companies, + properties={ + "employees": relationship(Employee, backref="company") + }, + ) mapper(Employee, employees, polymorphic_on=employees.c.type) - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') - mapper(Manager, inherits=Employee, polymorphic_identity='manager') + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") + mapper(Manager, inherits=Employee, polymorphic_identity="manager") s = create_session() - q1 = s.query(Engineer).\ - join(Engineer.company).\ - join(Manager, Company.employees) + q1 = ( + s.query(Engineer) + .join(Engineer.company) + .join(Manager, Company.employees) + ) - q2 = s.query(Engineer).\ - join(Engineer.company).\ - join(Manager, Company.company_id == Manager.company_id) + q2 = ( + s.query(Engineer) + .join(Engineer.company) + .join(Manager, Company.company_id == Manager.company_id) + ) - q3 = s.query(Engineer).\ - join(Engineer.company).\ - join(Manager, Company.employees.of_type(Manager)) + q3 = ( + s.query(Engineer) + .join(Engineer.company) + .join(Manager, Company.employees.of_type(Manager)) + ) - q4 = s.query(Engineer).\ - join(Company, Company.company_id == Engineer.company_id).\ - join(Manager, Company.employees.of_type(Manager)) + q4 = ( + s.query(Engineer) + .join(Company, Company.company_id == Engineer.company_id) + .join(Manager, Company.employees.of_type(Manager)) + ) - q5 = s.query(Engineer).\ - join(Company, Company.company_id == Engineer.company_id).\ - join(Manager, Company.company_id == Manager.company_id) + q5 = ( + s.query(Engineer) + .join(Company, Company.company_id == Engineer.company_id) + .join(Manager, Company.company_id == Manager.company_id) + ) # note that the query is incorrect SQL; we JOIN to # employees twice. However, this is what's expected so we seek @@ -877,38 +1022,49 @@ class RelationshipToSingleTest( "JOIN employees " "ON companies.company_id = employees.company_id " "AND employees.type IN (:type_1) " - "WHERE employees.type IN (:type_2)" + "WHERE employees.type IN (:type_2)", ) def test_relationship_to_subclass(self): - JuniorEngineer, Company, companies, Manager, \ - Employee, employees, Engineer = (self.classes.JuniorEngineer, - self.classes.Company, - self.tables.companies, - self.classes.Manager, - self.classes.Employee, - self.tables.employees, - self.classes.Engineer) - - mapper(Company, companies, properties={ - 'engineers': relationship(Engineer) - }) - mapper(Employee, employees, polymorphic_on=employees.c.type, - properties={'company': relationship(Company)}) - mapper(Manager, inherits=Employee, polymorphic_identity='manager') - mapper(Engineer, inherits=Employee, polymorphic_identity='engineer') - mapper(JuniorEngineer, inherits=Engineer, - polymorphic_identity='juniorengineer') + JuniorEngineer, Company, companies, Manager, Employee, employees, Engineer = ( + self.classes.JuniorEngineer, + self.classes.Company, + self.tables.companies, + self.classes.Manager, + self.classes.Employee, + self.tables.employees, + self.classes.Engineer, + ) + + mapper( + Company, + companies, + properties={"engineers": relationship(Engineer)}, + ) + mapper( + Employee, + employees, + polymorphic_on=employees.c.type, + properties={"company": relationship(Company)}, + ) + mapper(Manager, inherits=Employee, polymorphic_identity="manager") + mapper(Engineer, inherits=Employee, polymorphic_identity="engineer") + mapper( + JuniorEngineer, + inherits=Engineer, + polymorphic_identity="juniorengineer", + ) sess = sessionmaker()() - c1 = Company(name='c1') - c2 = Company(name='c2') + c1 = Company(name="c1") + c2 = Company(name="c2") - m1 = Manager(name='Tom', manager_data='data1', company=c1) - m2 = Manager(name='Tom2', manager_data='data2', company=c2) - e1 = Engineer(name='Kurt', engineer_info='knows how to hack', - company=c2) - e2 = JuniorEngineer(name='Ed', engineer_info='oh that ed', company=c1) + m1 = Manager(name="Tom", manager_data="data1", company=c1) + m2 = Manager(name="Tom2", manager_data="data2", company=c2) + e1 = Engineer( + name="Kurt", engineer_info="knows how to hack", company=c2 + ) + e2 = JuniorEngineer(name="Ed", engineer_info="oh that ed", company=c1) sess.add_all([c1, c2, m1, m2, e1, e2]) sess.commit() @@ -916,42 +1072,64 @@ class RelationshipToSingleTest( eq_(c2.engineers, [e1]) sess.expunge_all() - eq_(sess.query(Company).order_by(Company.name).all(), + eq_( + sess.query(Company).order_by(Company.name).all(), [ - Company(name='c1', engineers=[JuniorEngineer(name='Ed')]), - Company(name='c2', engineers=[Engineer(name='Kurt')])]) + Company(name="c1", engineers=[JuniorEngineer(name="Ed")]), + Company(name="c2", engineers=[Engineer(name="Kurt")]), + ], + ) # eager load join should limit to only "Engineer" sess.expunge_all() - eq_(sess.query(Company).options(joinedload('engineers')). - order_by(Company.name).all(), - [Company(name='c1', engineers=[JuniorEngineer(name='Ed')]), - Company(name='c2', engineers=[Engineer(name='Kurt')])]) + eq_( + sess.query(Company) + .options(joinedload("engineers")) + .order_by(Company.name) + .all(), + [ + Company(name="c1", engineers=[JuniorEngineer(name="Ed")]), + Company(name="c2", engineers=[Engineer(name="Kurt")]), + ], + ) # join() to Company.engineers, Employee as the requested entity sess.expunge_all() - eq_(sess.query(Company, Employee) + eq_( + sess.query(Company, Employee) .join(Company.engineers) .order_by(Company.name) .all(), - [(Company(name='c1'), JuniorEngineer(name='Ed')), - (Company(name='c2'), Engineer(name='Kurt'))]) + [ + (Company(name="c1"), JuniorEngineer(name="Ed")), + (Company(name="c2"), Engineer(name="Kurt")), + ], + ) # join() to Company.engineers, Engineer as the requested entity. # this actually applies the IN criterion twice which is less than # ideal. sess.expunge_all() - eq_(sess.query(Company, Engineer) + eq_( + sess.query(Company, Engineer) .join(Company.engineers) .order_by(Company.name) .all(), - [(Company(name='c1'), JuniorEngineer(name='Ed')), - (Company(name='c2'), Engineer(name='Kurt'))]) + [ + (Company(name="c1"), JuniorEngineer(name="Ed")), + (Company(name="c2"), Engineer(name="Kurt")), + ], + ) # join() to Company.engineers without any Employee/Engineer entity sess.expunge_all() - eq_(sess.query(Company).join(Company.engineers).filter( - Engineer.name.in_(['Tom', 'Kurt'])).all(), [Company(name='c2')]) + eq_( + sess.query(Company) + .join(Company.engineers) + .filter(Engineer.name.in_(["Tom", "Kurt"])) + .all(), + [Company(name="c2")], + ) # this however fails as it does not limit the subtypes to just # "Engineer". with joins constructed by filter(), we seem to be @@ -964,31 +1142,48 @@ class RelationshipToSingleTest( @testing.fails_on_everything_except() def go(): sess.expunge_all() - eq_(sess.query(Company).filter( - Company.company_id == Engineer.company_id).filter( - Engineer.name.in_(['Tom', 'Kurt'])).all(), - [Company(name='c2')]) + eq_( + sess.query(Company) + .filter(Company.company_id == Engineer.company_id) + .filter(Engineer.name.in_(["Tom", "Kurt"])) + .all(), + [Company(name="c2")], + ) + go() class ManyToManyToSingleTest(fixtures.MappedTest, AssertsCompiledSQL): - __dialect__ = 'default' + __dialect__ = "default" @classmethod def define_tables(cls, metadata): - Table('parent', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True)) - Table('m2m', metadata, - Column('parent_id', Integer, - ForeignKey('parent.id'), primary_key=True), - Column('child_id', Integer, - ForeignKey('child.id'), primary_key=True)) - Table('child', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('discriminator', String(20)), - Column('name', String(20))) + Table( + "parent", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + ) + Table( + "m2m", + metadata, + Column( + "parent_id", Integer, ForeignKey("parent.id"), primary_key=True + ), + Column( + "child_id", Integer, ForeignKey("child.id"), primary_key=True + ), + ) + Table( + "child", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("discriminator", String(20)), + Column("name", String(20)), + ) @classmethod def setup_classes(cls): @@ -1006,19 +1201,35 @@ class ManyToManyToSingleTest(fixtures.MappedTest, AssertsCompiledSQL): @classmethod def setup_mappers(cls): - mapper(cls.classes.Parent, cls.tables.parent, properties={ - "s1": relationship(cls.classes.SubChild1, - secondary=cls.tables.m2m, - uselist=False), - "s2": relationship(cls.classes.SubChild2, - secondary=cls.tables.m2m) - }) - mapper(cls.classes.Child, cls.tables.child, - polymorphic_on=cls.tables.child.c.discriminator) - mapper(cls.classes.SubChild1, inherits=cls.classes.Child, - polymorphic_identity='sub1') - mapper(cls.classes.SubChild2, inherits=cls.classes.Child, - polymorphic_identity='sub2') + mapper( + cls.classes.Parent, + cls.tables.parent, + properties={ + "s1": relationship( + cls.classes.SubChild1, + secondary=cls.tables.m2m, + uselist=False, + ), + "s2": relationship( + cls.classes.SubChild2, secondary=cls.tables.m2m + ), + }, + ) + mapper( + cls.classes.Child, + cls.tables.child, + polymorphic_on=cls.tables.child.c.discriminator, + ) + mapper( + cls.classes.SubChild1, + inherits=cls.classes.Child, + polymorphic_identity="sub1", + ) + mapper( + cls.classes.SubChild2, + inherits=cls.classes.Child, + polymorphic_identity="sub2", + ) @classmethod def insert_data(cls): @@ -1026,9 +1237,14 @@ class ManyToManyToSingleTest(fixtures.MappedTest, AssertsCompiledSQL): SubChild1 = cls.classes.SubChild1 SubChild2 = cls.classes.SubChild2 s = Session() - s.add_all([ - Parent(s1=SubChild1(name='sc1_1'), - s2=[SubChild2(name="sc2_1"), SubChild2(name="sc2_2")])]) + s.add_all( + [ + Parent( + s1=SubChild1(name="sc1_1"), + s2=[SubChild2(name="sc2_1"), SubChild2(name="sc2_2")], + ) + ] + ) s.commit() def test_eager_join(self): @@ -1038,7 +1254,7 @@ class ManyToManyToSingleTest(fixtures.MappedTest, AssertsCompiledSQL): s = Session() p1 = s.query(Parent).options(joinedload(Parent.s1)).all()[0] - eq_(p1.__dict__['s1'], SubChild1(name='sc1_1')) + eq_(p1.__dict__["s1"], SubChild1(name="sc1_1")) def test_manual_join(self): Parent = self.classes.Parent @@ -1048,7 +1264,7 @@ class ManyToManyToSingleTest(fixtures.MappedTest, AssertsCompiledSQL): s = Session() p1, c1 = s.query(Parent, Child).outerjoin(Parent.s1).all()[0] - eq_(c1, SubChild1(name='sc1_1')) + eq_(c1, SubChild1(name="sc1_1")) def test_assert_join_sql(self): Parent = self.classes.Parent @@ -1064,7 +1280,7 @@ class ManyToManyToSingleTest(fixtures.MappedTest, AssertsCompiledSQL): "FROM parent LEFT OUTER JOIN (m2m AS m2m_1 " "JOIN child ON child.id = m2m_1.child_id " "AND child.discriminator IN (:discriminator_1)) " - "ON parent.id = m2m_1.parent_id" + "ON parent.id = m2m_1.parent_id", ) def test_assert_joinedload_sql(self): @@ -1081,7 +1297,7 @@ class ManyToManyToSingleTest(fixtures.MappedTest, AssertsCompiledSQL): "FROM parent LEFT OUTER JOIN " "(m2m AS m2m_1 JOIN child AS child_1 " "ON child_1.id = m2m_1.child_id AND child_1.discriminator " - "IN (:discriminator_1)) ON parent.id = m2m_1.parent_id" + "IN (:discriminator_1)) ON parent.id = m2m_1.parent_id", ) @@ -1090,18 +1306,31 @@ class SingleOnJoinedTest(fixtures.MappedTest): def define_tables(cls, metadata): global persons_table, employees_table - persons_table = Table('persons', metadata, - Column('person_id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('name', String(50)), - Column('type', String(20), nullable=False)) + persons_table = Table( + "persons", + metadata, + Column( + "person_id", + Integer, + primary_key=True, + test_needs_autoincrement=True, + ), + Column("name", String(50)), + Column("type", String(20), nullable=False), + ) - employees_table = Table('employees', metadata, - Column('person_id', Integer, - ForeignKey('persons.person_id'), - primary_key=True), - Column('employee_data', String(50)), - Column('manager_data', String(50)),) + employees_table = Table( + "employees", + metadata, + Column( + "person_id", + Integer, + ForeignKey("persons.person_id"), + primary_key=True, + ), + Column("employee_data", String(50)), + Column("manager_data", String(50)), + ) def test_single_on_joined(self): class Person(fixtures.ComparableEntity): @@ -1113,44 +1342,67 @@ class SingleOnJoinedTest(fixtures.MappedTest): class Manager(Employee): pass - mapper(Person, persons_table, polymorphic_on=persons_table.c.type, - polymorphic_identity='person') - mapper(Employee, employees_table, inherits=Person, - polymorphic_identity='engineer') - mapper(Manager, inherits=Employee, polymorphic_identity='manager') + mapper( + Person, + persons_table, + polymorphic_on=persons_table.c.type, + polymorphic_identity="person", + ) + mapper( + Employee, + employees_table, + inherits=Person, + polymorphic_identity="engineer", + ) + mapper(Manager, inherits=Employee, polymorphic_identity="manager") sess = create_session() - sess.add(Person(name='p1')) - sess.add(Employee(name='e1', employee_data='ed1')) - sess.add(Manager(name='m1', employee_data='ed2', manager_data='md1')) + sess.add(Person(name="p1")) + sess.add(Employee(name="e1", employee_data="ed1")) + sess.add(Manager(name="m1", employee_data="ed2", manager_data="md1")) sess.flush() sess.expunge_all() - eq_(sess.query(Person).order_by(Person.person_id).all(), [ - Person(name='p1'), - Employee(name='e1', employee_data='ed1'), - Manager(name='m1', employee_data='ed2', manager_data='md1') - ]) + eq_( + sess.query(Person).order_by(Person.person_id).all(), + [ + Person(name="p1"), + Employee(name="e1", employee_data="ed1"), + Manager(name="m1", employee_data="ed2", manager_data="md1"), + ], + ) sess.expunge_all() - eq_(sess.query(Employee).order_by(Person.person_id).all(), [ - Employee(name='e1', employee_data='ed1'), - Manager(name='m1', employee_data='ed2', manager_data='md1') - ]) + eq_( + sess.query(Employee).order_by(Person.person_id).all(), + [ + Employee(name="e1", employee_data="ed1"), + Manager(name="m1", employee_data="ed2", manager_data="md1"), + ], + ) sess.expunge_all() - eq_(sess.query(Manager).order_by(Person.person_id).all(), [ - Manager(name='m1', employee_data='ed2', manager_data='md1') - ]) + eq_( + sess.query(Manager).order_by(Person.person_id).all(), + [Manager(name="m1", employee_data="ed2", manager_data="md1")], + ) sess.expunge_all() def go(): - eq_(sess.query(Person).with_polymorphic('*').order_by( - Person.person_id).all(), - [Person(name='p1'), - Employee(name='e1', employee_data='ed1'), - Manager( - name='m1', employee_data='ed2', manager_data='md1')]) + eq_( + sess.query(Person) + .with_polymorphic("*") + .order_by(Person.person_id) + .all(), + [ + Person(name="p1"), + Employee(name="e1", employee_data="ed1"), + Manager( + name="m1", employee_data="ed2", manager_data="md1" + ), + ], + ) + self.assert_sql_count(testing.db, go, 1) @@ -1162,7 +1414,8 @@ class EagerDefaultEvalTest(fixtures.DeclarativeMappedTest): class Foo(Base): __tablename__ = "foo" id = Column( - Integer, primary_key=True, test_needs_autoincrement=True) + Integer, primary_key=True, test_needs_autoincrement=True + ) type = Column(String(50)) created_at = Column(Integer, server_default="5") @@ -1170,7 +1423,7 @@ class EagerDefaultEvalTest(fixtures.DeclarativeMappedTest): "polymorphic_on": type, "polymorphic_identity": "foo", "eager_defaults": True, - "with_polymorphic": with_polymorphic + "with_polymorphic": with_polymorphic, } class Bar(Foo): @@ -1178,9 +1431,7 @@ class EagerDefaultEvalTest(fixtures.DeclarativeMappedTest): if include_sub_defaults: bat = Column(Integer, server_default="10") - __mapper_args__ = { - "polymorphic_identity": "bar", - } + __mapper_args__ = {"polymorphic_identity": "bar"} def test_persist_foo(self): Foo = self.classes.Foo @@ -1191,9 +1442,9 @@ class EagerDefaultEvalTest(fixtures.DeclarativeMappedTest): session.add(foo) session.flush() - eq_(foo.__dict__['created_at'], 5) + eq_(foo.__dict__["created_at"], 5) - assert 'bat' not in foo.__dict__ + assert "bat" not in foo.__dict__ session.close() @@ -1204,10 +1455,10 @@ class EagerDefaultEvalTest(fixtures.DeclarativeMappedTest): session.add(bar) session.flush() - eq_(bar.__dict__['created_at'], 5) + eq_(bar.__dict__["created_at"], 5) - if 'bat' in inspect(Bar).attrs: - eq_(bar.__dict__['bat'], 10) + if "bat" in inspect(Bar).attrs: + eq_(bar.__dict__["bat"], 10) session.close() @@ -1216,11 +1467,13 @@ class EagerDefaultEvalTestSubDefaults(EagerDefaultEvalTest): @classmethod def setup_classes(cls): super(EagerDefaultEvalTestSubDefaults, cls).setup_classes( - include_sub_defaults=True) + include_sub_defaults=True + ) class EagerDefaultEvalTestPolymorphic(EagerDefaultEvalTest): @classmethod def setup_classes(cls): super(EagerDefaultEvalTestPolymorphic, cls).setup_classes( - with_polymorphic="*") + with_polymorphic="*" + ) |
