diff options
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="*" + ) |